fix: (CDK) (Manifest) - Add Manifest Normalization module (reduce commonalities + handle schema $refs)

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1583,6 +1583,7 @@ definitions:
         title: URL Base
         description: The base URL (scheme and host, e.g. "https://api.example.com") to match.
         type: string
+        sharable: True
       url_path_pattern:
         title: URL Path Pattern
         description: A regular expression pattern to match the URL path.
@@ -1841,6 +1842,7 @@ definitions:
           - "{{ config['base_url'] or 'https://app.posthog.com'}}/api"
           - "https://connect.squareup.com/v2/quotes/{{ stream_partition['id'] }}/quote_line_groups"
           - "https://example.com/api/v1/resource/{{ next_page_token['id'] }}"
+        sharable: True
       path:
         title: URL Path
         description: Path the specific API endpoint that this stream represents. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.
@@ -1872,6 +1874,7 @@ definitions:
           - "$ref": "#/definitions/SessionTokenAuthenticator"
           - "$ref": "#/definitions/LegacySessionTokenAuthenticator"
           - "$ref": "#/definitions/SelectiveAuthenticator"
+        sharable: True
       error_handler:
         title: Error Handler
         description: Error handler component that defines how to handle errors.

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -39,6 +39,9 @@
 from airbyte_cdk.sources.declarative.parsers.manifest_component_transformer import (
     ManifestComponentTransformer,
 )
+from airbyte_cdk.sources.declarative.parsers.manifest_normalizer import (
+    ManifestNormalizer,
+)
 from airbyte_cdk.sources.declarative.parsers.manifest_reference_resolver import (
     ManifestReferenceResolver,
 )
@@ -57,6 +60,24 @@
 from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 
 
+def _get_declarative_component_schema() -> Dict[str, Any]:
+    try:
+        raw_component_schema = pkgutil.get_data(
+            "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml"
+        )
+        if raw_component_schema is not None:
+            declarative_component_schema = yaml.load(raw_component_schema, Loader=yaml.SafeLoader)
+            return declarative_component_schema  # type: ignore
+        else:
+            raise RuntimeError(
+                "Failed to read manifest component json schema required for deduplication"
+            )
+    except FileNotFoundError as e:
+        raise FileNotFoundError(
+            f"Failed to read manifest component json schema required for deduplication: {e}"
+        )
+
+
 class ManifestDeclarativeSource(DeclarativeSource):
     """Declarative source defined by a manifest of low-code components that define source connector behavior"""
 
@@ -78,6 +99,8 @@ def __init__(
             component_factory: optional factory if ModelToComponentFactory's default behavior needs to be tweaked.
         """
         self.logger = logging.getLogger(f"airbyte.{self.name}")
+
+        self._declarative_component_schema = _get_declarative_component_schema()
         # For ease of use we don't require the type to be specified at the top level manifest, but it should be included during processing
         manifest = dict(source_config)
         if "type" not in manifest:
@@ -87,6 +110,14 @@ def __init__(
         self.components_module: ModuleType | None = get_registered_components_module(config=config)
 
         resolved_source_config = ManifestReferenceResolver().preprocess_manifest(manifest)
+
+        if emit_connector_builder_messages:
+            # reduce commonalities in the manifest after the references have been resolved,
+            # used mostly for Connector Builder use cases.
+            resolved_source_config = ManifestNormalizer(
+                resolved_source_config, self._declarative_component_schema
+            ).normalize()
+
         propagated_source_config = ManifestComponentTransformer().propagate_types_and_parameters(
             "", resolved_source_config, {}
         )
@@ -266,22 +297,6 @@ def _validate_source(self) -> None:
         """
         Validates the connector manifest against the declarative component schema
         """
-        try:
-            raw_component_schema = pkgutil.get_data(
-                "airbyte_cdk", "sources/declarative/declarative_component_schema.yaml"
-            )
-            if raw_component_schema is not None:
-                declarative_component_schema = yaml.load(
-                    raw_component_schema, Loader=yaml.SafeLoader
-                )
-            else:
-                raise RuntimeError(
-                    "Failed to read manifest component json schema required for validation"
-                )
-        except FileNotFoundError as e:
-            raise FileNotFoundError(
-                f"Failed to read manifest component json schema required for validation: {e}"
-            )
 
         streams = self._source_config.get("streams")
         dynamic_streams = self._source_config.get("dynamic_streams")
@@ -291,7 +306,7 @@ def _validate_source(self) -> None:
             )
 
         try:
-            validate(self._source_config, declarative_component_schema)
+            validate(self._source_config, self._declarative_component_schema)
         except ValidationError as e:
             raise ValidationError(
                 "Validation against json schema defined in declarative_component_schema.yaml schema failed"

airbyte_cdk/sources/declarative/parsers/custom_exceptions.py

@@ -19,3 +19,12 @@ class UndefinedReferenceException(Exception):
 
     def __init__(self, path: str, reference: str) -> None:
         super().__init__(f"Undefined reference {reference} from {path}")
+
+
+class ManifestNormalizationException(Exception):
+    """
+    Raised when a circular reference is detected in a manifest.
+    """
+
+    def __init__(self, message: str) -> None:
+        super().__init__(f"Failed to deduplicate manifest: {message}")