Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added initial defaults objects in first step to oop #64

Draft
wants to merge 5 commits into
base: main
Choose a base branch
from
Draft

Added initial defaults objects in first step to oop #64

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2923dd4
Added initial defaults objects in first step to oop
allegra-noto Aug 15, 2024
fd291ee
Added metadata as a parameter object type
allegra-noto Aug 15, 2024
106000c
created get defaults function
allegra-noto Aug 19, 2024
c50fb21
Added type hints to output parameter lists
allegra-noto Aug 20, 2024
6a6d0cb
Fixed small errors
allegra-noto Aug 28, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
210 changes: 160 additions & 50 deletions google_cloud_automlops/orchestration/base.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,14 @@
DEFAULT_PIPELINE_NAME,
GENERATED_DEFAULTS_FILE
)
from google_cloud_automlops.utils.enums import (
GCP,
PipelineSpecs,
Tooling,
Defaults,
GCPLocations,
Parameter
)

T = TypeVar('T')

Expand Down Expand Up @@ -78,10 +86,7 @@ def __init__(self,
self.src_code = get_function_source_definition(self.func)

# Instantiate attributes to be set during build
self.artifact_repo_location = None
self.artifact_repo_name = None
self.project_id = None
self.naming_prefix = None
self.defaults = None

def build(self):
"""Instantiates an abstract built method to create and write task files. Also reads in
Expand All @@ -92,12 +97,50 @@ def build(self):
"""

defaults = read_yaml_file(GENERATED_DEFAULTS_FILE)
self.artifact_repo_location = defaults['gcp']['artifact_repo_location']
self.artifact_repo_name = defaults['gcp']['artifact_repo_name']
self.project_id = defaults['gcp']['project_id']
self.naming_prefix = defaults['gcp']['naming_prefix']

raise NotImplementedError('Subclass needs to define this.')
gcp = GCP(
artifact_repo_location = GCPLocations(defaults["gcp"]["artifact_repo_location"]),
artifact_repo_name = defaults["gcp"]["artifact_repo_name"],
artifact_repo_type = defaults["gcp"]["artifact_repo_type"],
base_image = defaults["gcp"]["base_image"],
build_trigger_location = defaults["gcp"]["build_trigger_location"],
build_trigger_name = defaults["gcp"]["build_trigger_name"],
naming_prefix = defaults["gcp"]["naming_prefix"],
pipeline_job_runner_service_account = defaults["gcp"]["pipeline_job_runner_service_account"],
pipeline_job_submission_service_location = defaults["gcp"]["pipeline_job_submission_service_location"],
pipeline_job_submission_service_name = defaults["gcp"]["pipeline_job_submission_service_name"],
pipeline_job_submission_service_type = defaults["gcp"]["pipeline_job_submission_service_type"],
project_id = defaults["gcp"]["project_id"],
setup_model_monitoring = defaults["gcp"]["setup_model_monitoring"],
pubsub_topic_name = defaults["gcp"]["pubsub_topic_name"],
schedule_location = defaults["gcp"]["schedule_location"],
schedule_name = defaults["gcp"]["schedule_name"],
schedule_pattern = defaults["gcp"]["schedule_pattern"],
source_repository_branch = defaults["gcp"]["source_repository_branch"],
source_repository_name = defaults["gcp"]["source_repository_name"],
source_repository_type = defaults["gcp"]["source_repository_type"],
storage_bucket_location = defaults["gcp"]["storage_bucket_location"],
storage_bucket_name = defaults["gcp"]["storage_bucket_name"],
vpc_connector = defaults["gcp"]["vpc_connector"],
)
pipeline_specs = PipelineSpecs(
gs_pipeline_job_spec_path = defaults["pipelines"]["gs_pipeline_job_spec_path"],
parameter_values_path = defaults["pipelines"]["parameter_values_path"],
pipeline_component_directory = defaults["pipelines"]["pipeline_component_directory"],
pipeline_job_spec_path = defaults["pipelines"]["pipeline_job_spec_path"],
pipeline_region = defaults["pipelines"]["pipeline_region"],
pipeline_storage_path = defaults["pipelines"]["pipeline_storage_path"],
)
tooling = Tooling(
deployment_framework = defaults["tooling"]["deployment_framework"],
provisioning_framework = defaults["tooling"]["provisioning_framework"],
orchestration_framework = defaults["tooling"]["orchestration_framework"],
use_ci = defaults["tooling"]["use_ci"],
)
self.defaults = Defaults(
gcp = gcp,
pipeline_specs = pipeline_specs,
tooling = tooling
)

def _get_function_return_types(self) -> list:
"""Returns a formatted list of function return types.
Expand All @@ -124,14 +167,15 @@ def _get_function_return_types(self) -> list:
if not (hasattr(annotation,'__annotations__') and isinstance(annotation.__annotations__, dict)):
raise TypeError(f'''Return type hint for function "{self.name}" must be a NamedTuple.''')

# Creates a dictionary of metadata for each object returned by component
# Creates a parameter object for each parameter returned by component
outputs = []
allegra-noto marked this conversation as resolved.
Show resolved Hide resolved
for name, type_ in annotation.__annotations__.items():
metadata = {}
metadata['name'] = name
metadata['type'] = type_
metadata['description'] = None
outputs.append(metadata)
p = Parameter(
name=name,
type=type_,
description=None
)
outputs.append(p)
return outputs

def _get_function_parameters(self) -> list:
Expand All @@ -152,16 +196,17 @@ def _get_function_parameters(self) -> list:
# Extract parameter metadata
parameter_holder = []
for param in parameters:
metadata = {}
metadata['name'] = param.name
metadata['description'] = doc_dict.get(param.name)
metadata['type'] = self.maybe_strip_optional_from_annotation(
param.annotation)
parameter_holder.append(metadata)
p = Parameter(
name=param.name,
type=self.maybe_strip_optional_from_annotation(
param.annotation),
description=doc_dict.get(param.name)
)
parameter_holder.append(p)
# pylint: disable=protected-access
if metadata['type'] == inspect._empty:
if p.type == inspect._empty:
raise TypeError(
f'''Missing type hint for parameter "{metadata['name']}". '''
f'''Missing type hint for parameter "{p.name}". '''
f'''Please specify the type for this parameter.''')
return parameter_holder

Expand Down Expand Up @@ -216,14 +261,9 @@ def __init__(self,
self.comps = self.get_pipeline_components(func, comps_dict)

# Instantiate attributes to be set at build process
self.base_image = None
self.defaults = None
self.custom_training_job_specs = None
self.pipeline_params = None
self.pubsub_topic_name = None
self.use_ci = None
self.project_id = None
self.gs_pipeline_job_spec_path = None
self.setup_model_monitoring = None

def build(self,
pipeline_params: dict,
Expand All @@ -250,14 +290,50 @@ def build(self,

# Extract additional attributes from defaults file
defaults = read_yaml_file(GENERATED_DEFAULTS_FILE)
self.project_id = defaults['gcp']['project_id']
self.gs_pipeline_job_spec_path = defaults['pipelines']['gs_pipeline_job_spec_path']
self.base_image = defaults['gcp']['base_image']
self.pubsub_topic_name = defaults['gcp']['pubsub_topic_name']
self.use_ci = defaults['tooling']['use_ci']
self.setup_model_monitoring = defaults['gcp']['setup_model_monitoring']

raise NotImplementedError('Subclass needs to define this.')
gcp = GCP(
artifact_repo_location = GCPLocations(defaults["gcp"]["artifact_repo_location"]),
artifact_repo_name = defaults["gcp"]["artifact_repo_name"],
artifact_repo_type = defaults["gcp"]["artifact_repo_type"],
base_image = defaults["gcp"]["base_image"],
build_trigger_location = defaults["gcp"]["build_trigger_location"],
build_trigger_name = defaults["gcp"]["build_trigger_name"],
naming_prefix = defaults["gcp"]["naming_prefix"],
pipeline_job_runner_service_account = defaults["gcp"]["pipeline_job_runner_service_account"],
pipeline_job_submission_service_location = defaults["gcp"]["pipeline_job_submission_service_location"],
pipeline_job_submission_service_name = defaults["gcp"]["pipeline_job_submission_service_name"],
pipeline_job_submission_service_type = defaults["gcp"]["pipeline_job_submission_service_type"],
project_id = defaults["gcp"]["project_id"],
setup_model_monitoring = defaults["gcp"]["setup_model_monitoring"],
pubsub_topic_name = defaults["gcp"]["pubsub_topic_name"],
schedule_location = defaults["gcp"]["schedule_location"],
schedule_name = defaults["gcp"]["schedule_name"],
schedule_pattern = defaults["gcp"]["schedule_pattern"],
source_repository_branch = defaults["gcp"]["source_repository_branch"],
source_repository_name = defaults["gcp"]["source_repository_name"],
source_repository_type = defaults["gcp"]["source_repository_type"],
storage_bucket_location = defaults["gcp"]["storage_bucket_location"],
storage_bucket_name = defaults["gcp"]["storage_bucket_name"],
vpc_connector = defaults["gcp"]["vpc_connector"],
)
pipeline_specs = PipelineSpecs(
gs_pipeline_job_spec_path = defaults["pipelines"]["gs_pipeline_job_spec_path"],
parameter_values_path = defaults["pipelines"]["parameter_values_path"],
pipeline_component_directory = defaults["pipelines"]["pipeline_component_directory"],
pipeline_job_spec_path = defaults["pipelines"]["pipeline_job_spec_path"],
pipeline_region = defaults["pipelines"]["pipeline_region"],
pipeline_storage_path = defaults["pipelines"]["pipeline_storage_path"],
)
tooling = Tooling(
deployment_framework = defaults["tooling"]["deployment_framework"],
provisioning_framework = defaults["tooling"]["provisioning_framework"],
orchestration_framework = defaults["tooling"]["orchestration_framework"],
use_ci = defaults["tooling"]["use_ci"],
)
self.defaults = Defaults(
gcp = gcp,
pipeline_specs = pipeline_specs,
tooling = tooling
)

def get_pipeline_components(self,
pipeline_func: Callable,
Expand Down Expand Up @@ -301,12 +377,7 @@ class BaseServices():
def __init__(self) -> None:
"""Instantiates a generic Services object.
"""
self.pipeline_storage_path = None
self.pipeline_job_runner_service_account = None
self.pipeline_job_submission_service_type = None
self.project_id = None
self.pipeline_job_submission_service_type = None
self.setup_model_monitoring = None
self.defaults = None

# Set directory for files to be written to
self.submission_service_base_dir = BASE_DIR + 'services/submission_service'
Expand All @@ -326,11 +397,50 @@ def build(self):
"""
# Extract additional attributes from defaults file
defaults = read_yaml_file(GENERATED_DEFAULTS_FILE)
self.pipeline_storage_path = defaults['pipelines']['pipeline_storage_path']
self.pipeline_job_runner_service_account = defaults['gcp']['pipeline_job_runner_service_account']
self.pipeline_job_submission_service_type = defaults['gcp']['pipeline_job_submission_service_type']
self.project_id = defaults['gcp']['project_id']
self.setup_model_monitoring = defaults['gcp']['setup_model_monitoring']
gcp = GCP(
allegra-noto marked this conversation as resolved.
Show resolved Hide resolved
artifact_repo_location = GCPLocations(defaults["gcp"]["artifact_repo_location"]),
artifact_repo_name = defaults["gcp"]["artifact_repo_name"],
artifact_repo_type = defaults["gcp"]["artifact_repo_type"],
base_image = defaults["gcp"]["base_image"],
build_trigger_location = defaults["gcp"]["build_trigger_location"],
build_trigger_name = defaults["gcp"]["build_trigger_name"],
naming_prefix = defaults["gcp"]["naming_prefix"],
pipeline_job_runner_service_account = defaults["gcp"]["pipeline_job_runner_service_account"],
pipeline_job_submission_service_location = defaults["gcp"]["pipeline_job_submission_service_location"],
pipeline_job_submission_service_name = defaults["gcp"]["pipeline_job_submission_service_name"],
pipeline_job_submission_service_type = defaults["gcp"]["pipeline_job_submission_service_type"],
project_id = defaults["gcp"]["project_id"],
setup_model_monitoring = defaults["gcp"]["setup_model_monitoring"],
pubsub_topic_name = defaults["gcp"]["pubsub_topic_name"],
schedule_location = defaults["gcp"]["schedule_location"],
schedule_name = defaults["gcp"]["schedule_name"],
schedule_pattern = defaults["gcp"]["schedule_pattern"],
source_repository_branch = defaults["gcp"]["source_repository_branch"],
source_repository_name = defaults["gcp"]["source_repository_name"],
source_repository_type = defaults["gcp"]["source_repository_type"],
storage_bucket_location = defaults["gcp"]["storage_bucket_location"],
storage_bucket_name = defaults["gcp"]["storage_bucket_name"],
vpc_connector = defaults["gcp"]["vpc_connector"],
)
pipeline_specs = PipelineSpecs(
gs_pipeline_job_spec_path = defaults["pipelines"]["gs_pipeline_job_spec_path"],
parameter_values_path = defaults["pipelines"]["parameter_values_path"],
pipeline_component_directory = defaults["pipelines"]["pipeline_component_directory"],
pipeline_job_spec_path = defaults["pipelines"]["pipeline_job_spec_path"],
pipeline_region = defaults["pipelines"]["pipeline_region"],
pipeline_storage_path = defaults["pipelines"]["pipeline_storage_path"],
)
tooling = Tooling(
deployment_framework = defaults["tooling"]["deployment_framework"],
provisioning_framework = defaults["tooling"]["provisioning_framework"],
orchestration_framework = defaults["tooling"]["orchestration_framework"],
use_ci = defaults["tooling"]["use_ci"],
)
self.defaults = Defaults(
gcp = gcp,
pipeline_specs = pipeline_specs,
tooling = tooling
)

# Set directory for files to be written to
self.submission_service_base_dir = BASE_DIR + 'services/submission_service'
Expand All @@ -339,7 +449,7 @@ def build(self):
self._build_submission_services()

# Setup model monitoring
if self.setup_model_monitoring:
if self.defaults.gcp.setup_model_monitoring:
self._build_monitoring()

def _build_monitoring(self):
Expand Down
Loading
Loading