Fix tests and format

Author: bnchrch

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/__init__.py

@@ -3,7 +3,6 @@
 #
 from dagster import Definitions
 
-from orchestrator.jobs.registry import generate_registry, generate_registry_markdown, generate_local_metadata_files
 from orchestrator.resources.gcp import gcp_gcs_client, gcs_bucket_manager, gcs_directory_blobs, gcs_file_blob, gcs_file_manager
 from orchestrator.resources.github import github_client, github_connector_repo, github_connectors_directory
 from orchestrator.resources.local import simple_local_file_manager
@@ -48,6 +47,8 @@
  cloud_registry_diff_dataframe,
  oss_registry_diff_dataframe,
  metadata_directory_report,
+ oss_registry_diff_report,
+ cloud_registry_diff_report,
 )
 
 from orchestrator.jobs.registry import generate_registry_markdown, generate_local_metadata_files, generate_registry
@@ -64,6 +65,7 @@
  cached_specs,
  cloud_destinations_dataframe,
  cloud_registry_diff_dataframe,
+ cloud_registry_diff_report,
  cloud_registry_diff,
  cloud_registry_from_metadata,
  cloud_sources_dataframe,
@@ -83,6 +85,7 @@
  metadata_directory_report,
  oss_destinations_dataframe,
  oss_registry_diff_dataframe,
+ oss_registry_diff_report,
  oss_registry_diff,
  oss_registry_from_metadata,
  oss_sources_dataframe,
@@ -105,12 +108,9 @@
  "gcs_bucket_manager": gcs_bucket_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}}),
  "registry_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRIES_FOLDER}),
  "registry_report_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REPORT_FOLDER}),
- "registry_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRIES_FOLDER}),
  "metadata_file_blobs": gcs_directory_blobs.configured({"prefix": METADATA_FOLDER, "suffix": METADATA_FILE_NAME}),
-
  "legacy_oss_registry_gcs_blob": gcs_file_blob.configured({"prefix": "", "gcs_filename": "oss_catalog.json"}),
  "legacy_cloud_registry_gcs_blob": gcs_file_blob.configured({"prefix": "", "gcs_filename": "cloud_catalog.json"}),
-
  "latest_oss_registry_gcs_blob": gcs_file_blob.configured({"prefix": REGISTRIES_FOLDER, "gcs_filename": "oss_registry.json"}),
  "latest_cloud_registry_gcs_blob": gcs_file_blob.configured({"prefix": REGISTRIES_FOLDER, "gcs_filename": "cloud_registry.json"}),
 }

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/assets/dev.py

@@ -3,10 +3,12 @@
 import json
 from pydash.collections import key_by
 from deepdiff import DeepDiff
-from dagster import Output, asset, OpExecutionContext
+from dagster import Output, asset, OpExecutionContext, MetadataValue
 from typing import List
+
 from orchestrator.utils.dagster_helpers import OutputDataFrame, output_dataframe
 from orchestrator.models.metadata import PartialMetadataDefinition
+
 from metadata_service.models.generated.ConnectorRegistryV0 import ConnectorRegistryV0
 
 
@@ -264,9 +266,37 @@ def oss_registry_diff_dataframe(oss_registry_diff: dict) -> OutputDataFrame:
 
 
 @asset(required_resource_keys={"metadata_file_blobs"}, group_name=GROUP_NAME)
-def metadata_directory_report(context):
+def metadata_directory_report(context: OpExecutionContext):
  metadata_file_blobs = context.resources.metadata_file_blobs
  blobs = [blob.name for blob in metadata_file_blobs if blob.name.endswith("metadata.yaml")]
  blobs_df = pd.DataFrame(blobs)
 
  return output_dataframe(blobs_df)
+
+
+@asset(required_resource_keys={"registry_report_directory_manager"}, group_name=GROUP_NAME)
+def oss_registry_diff_report(context: OpExecutionContext, oss_registry_diff_dataframe: pd.DataFrame):
+ markdown = oss_registry_diff_dataframe.to_markdown()
+
+ registry_report_directory_manager = context.resources.registry_report_directory_manager
+ file_handle = registry_report_directory_manager.write_data(markdown.encode(), ext="md", key="dev/oss_registry_diff_report")
+
+ metadata = {
+ "preview": MetadataValue.md(markdown),
+ "gcs_path": MetadataValue.url(file_handle.gcs_path),
+ }
+ return Output(metadata=metadata, value=file_handle)
+
+
+@asset(required_resource_keys={"registry_report_directory_manager"}, group_name=GROUP_NAME)
+def cloud_registry_diff_report(context: OpExecutionContext, cloud_registry_diff_dataframe: pd.DataFrame):
+ markdown = cloud_registry_diff_dataframe.to_markdown()
+
+ registry_report_directory_manager = context.resources.registry_report_directory_manager
+ file_handle = registry_report_directory_manager.write_data(markdown.encode(), ext="md", key="dev/cloud_registry_diff_report")
+
+ metadata = {
+ "preview": MetadataValue.md(markdown),
+ "gcs_path": MetadataValue.url(file_handle.gcs_path),
+ }
+ return Output(metadata=metadata, value=file_handle)

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/assets/metadata.py

@@ -202,7 +202,9 @@ def valid_metadata_report_dataframe(overrode_metadata_definitions: List[PartialM
 def legacy_registry_derived_metadata_definitions(
  legacy_cloud_sources_dataframe, legacy_cloud_destinations_dataframe, legacy_oss_sources_dataframe, legacy_oss_destinations_dataframe
 ) -> Output[List[PartialMetadataDefinition]]:
- sources_metadata_list = merge_into_metadata_definitions("sourceDefinitionId", "source", legacy_oss_sources_dataframe, legacy_cloud_sources_dataframe)
+ sources_metadata_list = merge_into_metadata_definitions(
+ "sourceDefinitionId", "source", legacy_oss_sources_dataframe, legacy_cloud_sources_dataframe
+ )
  destinations_metadata_list = merge_into_metadata_definitions(
  "destinationDefinitionId", "destination", legacy_oss_destinations_dataframe, legacy_cloud_destinations_dataframe
  )

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/assets/registry.py

@@ -19,7 +19,6 @@
 from metadata_service.models.generated.ConnectorRegistryV0 import ConnectorRegistryV0
 
 
-
 GROUP_NAME = "registry"
 
 # ERRORS
@@ -112,7 +111,9 @@ def is_metadata_connector_type(metadata_definition: dict, connector_type: str) -
  return metadata_definition["data"]["connectorType"] == connector_type
 
 
-def construct_registry_from_metadata(legacy_registry_derived_metadata_definitions: List[MetadataDefinition], registry_name: str) -> ConnectorRegistryV0:
+def construct_registry_from_metadata(
+ legacy_registry_derived_metadata_definitions: List[MetadataDefinition], registry_name: str
+) -> ConnectorRegistryV0:
  """Construct the registry from the metadata definitions.
 
  Args:
@@ -136,6 +137,7 @@ def construct_registry_from_metadata(legacy_registry_derived_metadata_definition
 
  return {"sources": registry_sources, "destinations": registry_destinations}
 
+
 def construct_registry_with_spec_from_registry(registry: dict, cached_specs: OutputDataFrame) -> dict:
  entries = [("source", entry) for entry in registry["sources"]] + [("destinations", entry) for entry in registry["destinations"]]
 
@@ -157,7 +159,10 @@ def construct_registry_with_spec_from_registry(registry: dict, cached_specs: Out
  raise MissingCachedSpecError(f"No cached spec found for {entry['dockerRegistry']:{entry['dockerImageTag']}}")
  return registry_with_specs
 
-def persist_registry_to_json(registry: ConnectorRegistryV0, registry_name: str, registry_directory_manager: GCSFileManager) -> GCSFileHandle:
+
+def persist_registry_to_json(
+ registry: ConnectorRegistryV0, registry_name: str, registry_directory_manager: GCSFileManager
+) -> GCSFileHandle:
  """Persist the registry to a json file.
 
  Args:
@@ -173,10 +178,14 @@ def persist_registry_to_json(registry: ConnectorRegistryV0, registry_name: str,
  file_handle = registry_directory_manager.write_data(registry_json.encode("utf-8"), ext="json", key=registry_file_name)
  return file_handle
 
+
 # New Registry
 
+
 @asset(required_resource_keys={"registry_directory_manager"}, group_name=GROUP_NAME)
-def cloud_registry_from_metadata(context: OpExecutionContext, metadata_definitions: List[MetadataDefinition], cached_specs: OutputDataFrame) -> Output[ConnectorRegistryV0]:
+def cloud_registry_from_metadata(
+ context: OpExecutionContext, metadata_definitions: List[MetadataDefinition], cached_specs: OutputDataFrame
+) -> Output[ConnectorRegistryV0]:
  """
  This asset is used to generate the cloud registry from the metadata definitions.
  """
@@ -185,19 +194,21 @@ def cloud_registry_from_metadata(context: OpExecutionContext, metadata_definitio
 
  from_metadata = construct_registry_from_metadata(metadata_definitions, registry_name)
  registry_dict = construct_registry_with_spec_from_registry(from_metadata, cached_specs)
- reigstry_model = ConnectorRegistryV0.parse_obj(registry_dict)
+ registry_model = ConnectorRegistryV0.parse_obj(registry_dict)
 
- file_handle = persist_registry_to_json(reigstry_model, registry_name, registry_directory_manager)
+ file_handle = persist_registry_to_json(registry_model, registry_name, registry_directory_manager)
 
  metadata = {
  "gcs_path": MetadataValue.url(file_handle.gcs_path),
  }
 
- return Output(metadata=metadata, value=reigstry_model)
+ return Output(metadata=metadata, value=registry_model)
 
 
 @asset(required_resource_keys={"registry_directory_manager"}, group_name=GROUP_NAME)
-def oss_registry_from_metadata(context: OpExecutionContext, metadata_definitions: List[MetadataDefinition], cached_specs: OutputDataFrame) -> Output[ConnectorRegistryV0]:
+def oss_registry_from_metadata(
+ context: OpExecutionContext, metadata_definitions: List[MetadataDefinition], cached_specs: OutputDataFrame
+) -> Output[ConnectorRegistryV0]:
  """
  This asset is used to generate the oss registry from the metadata definitions.
  """
@@ -206,15 +217,16 @@ def oss_registry_from_metadata(context: OpExecutionContext, metadata_definitions
 
  from_metadata = construct_registry_from_metadata(metadata_definitions, registry_name)
  registry_dict = construct_registry_with_spec_from_registry(from_metadata, cached_specs)
- reigstry_model = ConnectorRegistryV0.parse_obj(registry_dict)
+ registry_model = ConnectorRegistryV0.parse_obj(registry_dict)
 
- file_handle = persist_registry_to_json(reigstry_model, registry_name, registry_directory_manager)
+ file_handle = persist_registry_to_json(registry_model, registry_name, registry_directory_manager)
 
  metadata = {
  "gcs_path": MetadataValue.url(file_handle.gcs_path),
  }
 
- return Output(metadata=metadata, value=reigstry_model)
+ return Output(metadata=metadata, value=registry_model)
+
 
 @asset(group_name=GROUP_NAME)
 def cloud_sources_dataframe(cloud_registry_from_metadata: ConnectorRegistryV0) -> OutputDataFrame:
@@ -243,8 +255,10 @@ def oss_destinations_dataframe(oss_registry_from_metadata: ConnectorRegistryV0)
  destinations = oss_registry_from_metadata_dict["destinations"]
  return output_dataframe(pd.DataFrame(destinations))
 
+
 # Old Registry
 
+
 @asset(group_name=GROUP_NAME)
 def legacy_cloud_sources_dataframe(legacy_cloud_registry_dict: dict) -> OutputDataFrame:
  sources = legacy_cloud_registry_dict["sources"]
@@ -278,6 +292,7 @@ def legacy_cloud_registry(legacy_cloud_registry_dict: dict) -> ConnectorRegistry
 def legacy_oss_registry(legacy_oss_registry_dict: dict) -> ConnectorRegistryV0:
  return ConnectorRegistryV0.parse_obj(legacy_oss_registry_dict)
 
+
 @asset(required_resource_keys={"legacy_cloud_registry_gcs_blob"}, group_name=GROUP_NAME)
 def legacy_cloud_registry_dict(context: OpExecutionContext) -> dict:
  oss_registry_file = context.resources.legacy_cloud_registry_gcs_blob
@@ -292,4 +307,3 @@ def legacy_oss_registry_dict(context: OpExecutionContext) -> dict:
  json_string = oss_registry_file.download_as_string().decode("utf-8")
  oss_registry_dict = json.loads(json_string)
  return oss_registry_dict
-

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/jobs/registry.py

@@ -1,6 +1,8 @@
 from dagster import define_asset_job, AssetSelection
 
-registries_inclusive = AssetSelection.keys("metadata_directory_report", "cloud_registry_from_metadata", "oss_registry_from_metadata").upstream()
+registries_inclusive = AssetSelection.keys(
+ "metadata_directory_report", "cloud_registry_from_metadata", "oss_registry_from_metadata"
+).upstream()
 registry_reports_inclusive = AssetSelection.keys("connector_registry_location_html", "connector_registry_location_markdown").upstream()
 
 generate_registry = define_asset_job(name="generate_registry", selection=registries_inclusive)

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/models/metadata.py

@@ -25,6 +25,7 @@ def is_valid(self) -> Tuple[bool, Optional[Any]]:
  except ValidationError as e:
  return (False, e)
 
+
 class PydanitcDictMixin:
  def __getitem__(self, key: str):
  return self.__dict__[key]
@@ -36,5 +37,6 @@ def __setitem__(self, key: str, value: Any):
 class PartialMetadataDefinition(PydanticDelayValidationMixin, PydanitcDictMixin, ConnectorMetadataDefinitionV0):
  pass
 
+
 class MetadataDefinition(PydanitcDictMixin, ConnectorMetadataDefinitionV0):
  pass

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/resources/gcp.py

@@ -37,10 +37,7 @@ def gcs_bucket_manager(resource_context: InitResourceContext) -> storage.Bucket:
 
 @resource(
  required_resource_keys={"gcp_gcs_client"},
- config_schema={
- "gcs_bucket": StringSource,
- "prefix": StringSource
- },
+ config_schema={"gcs_bucket": StringSource, "prefix": StringSource},
 )
 def gcs_file_manager(resource_context) -> GCSFileManager:
  """FileManager that provides abstract access to GCS.

airbyte-ci/connectors/metadata_service/orchestrator/orchestrator/utils/object_helpers.py

@@ -31,6 +31,7 @@ def f(*args, **kwargs):
 
  return f
 
+
 def to_json_sanitized_dict(pydantic_model_obj: BaseModel) -> dict:
  """A helper function to convert a pydantic model to a sanitized dict.
 

airbyte-ci/connectors/metadata_service/orchestrator/tests/test_debug.py

@@ -10,10 +10,8 @@
  cloud_sources_dataframe,
  oss_registry_from_metadata,
  cloud_registry_from_metadata,
- cloud_registry_from_metadata,
- oss_registry_from_metadata,
  legacy_oss_registry_dict,
- legacy_oss_registry
+ legacy_oss_registry,
 )
 from orchestrator.assets.metadata import (
  legacy_registry_derived_metadata_definitions,
@@ -64,7 +62,9 @@ def debug_registry_projection():
  oss_sources_df = oss_sources_dataframe(oss_registry_dict).value
  # github_connector_folders_list = github_connector_folders(context).value
 
- legacy_registry_derived_metadata_definitions(context, cloud_sources_df, cloud_destinations_df, oss_sources_df, oss_destinations_df).value
+ legacy_registry_derived_metadata_definitions(
+ context, cloud_sources_df, cloud_destinations_df, oss_sources_df, oss_destinations_df
+ ).value
  # valid_metadata_report_dataframe_df = valid_metadata_report_dataframe(metadata_definitions_df).value
 
  # all_sources_df = all_sources_dataframe(cloud_sources_df, oss_sources_df, github_connector_folders_list, valid_metadata_report_dataframe_df)
@@ -73,6 +73,7 @@ def debug_registry_projection():
  # connector_registry_location_html(context, all_sources_df, all_destinations_df)
  # connector_registry_location_markdown(context, all_sources_df, all_destinations_df)
 
+
 def debug_registry_generation():
  resources = {
  "gcp_gcs_client": gcp_gcs_client.configured(
@@ -81,7 +82,7 @@ def debug_registry_generation():
  }
  ),
  "gcs_bucket_manager": gcs_bucket_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}}),
- "registry_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRY_FOLDER}),
+ "registry_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRIES_FOLDER}),
  "metadata_file_blobs": gcs_directory_blobs.configured({"prefix": METADATA_FOLDER, "suffix": METADATA_FILE_NAME}),
  }
 
@@ -90,15 +91,16 @@ def debug_registry_generation():
  oss_registry_from_metadata(context, metadata_definitions_asset).value
  # cloud_registry_from_metadata(context, metadata_definitions_asset).value
 
-def test_debug_registry_diff():
+
+def debug_registry_diff():
  resources = {
  "gcp_gcs_client": gcp_gcs_client.configured(
  {
  "gcp_gcs_cred_string": {"env": "GCS_CREDENTIALS"},
  }
  ),
  "gcs_bucket_manager": gcs_bucket_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}}),
- "registry_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRY_FOLDER}),
+ "registry_directory_manager": gcs_file_manager.configured({"gcs_bucket": {"env": "METADATA_BUCKET"}, "prefix": REGISTRIES_FOLDER}),
  "metadata_file_blobs": gcs_directory_blobs.configured({"prefix": METADATA_FOLDER, "suffix": METADATA_FILE_NAME}),
  "legacy_oss_registry_gcs_blob": gcs_file_blob.configured({"prefix": "", "gcs_filename": "oss_catalog.json"}),
  }