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

Author: Baz

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -56,12 +56,22 @@ def get_limits(config: Mapping[str, Any]) -> TestLimits:
     return TestLimits(max_records, max_pages_per_slice, max_slices, max_streams)
 
 
+def should_normalize_manifest(config: Mapping[str, Any]) -> bool:
+    """
+    Check if the manifest should be normalized.
+    :param config: The configuration to check
+    :return: True if the manifest should be normalized, False otherwise.
+    """
+    return config.get("__should_normalize", False)
+
+
 def create_source(config: Mapping[str, Any], limits: TestLimits) -> ManifestDeclarativeSource:
     manifest = config["__injected_declarative_manifest"]
     return ManifestDeclarativeSource(
         config=config,
         emit_connector_builder_messages=True,
         source_config=manifest,
+        normalize_manifest=should_normalize_manifest(config),
         component_factory=ModelToComponentFactory(
             emit_connector_builder_messages=True,
             limit_pages_fetched_per_slice=limits.max_pages_per_slice,

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1917,8 +1917,9 @@ definitions:
         type: string
         enum: [HttpRequester]
       url_base:
+        linkable: true
         title: API Base URL
-        description: Base URL of the API source. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.
+        description: The Base URL of the API source. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.
         type: string
         interpolation_context:
           - config
@@ -1936,7 +1937,7 @@ definitions:
           - "https://example.com/api/v1/resource/{{ next_page_token['id'] }}"
       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.
+        description: The 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.
         type: string
         interpolation_context:
           - config
@@ -1964,6 +1965,7 @@ definitions:
           - POST
       authenticator:
         title: Authenticator
+        linkable: true
         description: Authentication method to use for requests sent to the API.
         anyOf:
           - "$ref": "#/definitions/NoAuth"

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -29,7 +29,9 @@
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     DeclarativeStream as DeclarativeStreamModel,
 )
-from airbyte_cdk.sources.declarative.models.declarative_component_schema import Spec as SpecModel
+from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
+    Spec as SpecModel,
+)
 from airbyte_cdk.sources.declarative.models.declarative_component_schema import (
     StateDelegatingStream as StateDelegatingStreamModel,
 )
@@ -39,6 +41,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 +62,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"""
 
@@ -68,29 +91,25 @@ def __init__(
         debug: bool = False,
         emit_connector_builder_messages: bool = False,
         component_factory: Optional[ModelToComponentFactory] = None,
-    ):
+        normalize_manifest: Optional[bool] = False,
+    ) -> None:
         """
         Args:
             config: The provided config dict.
             source_config: The manifest of low-code components that describe the source connector.
             debug: True if debug mode is enabled.
             emit_connector_builder_messages: True if messages should be emitted to the connector builder.
             component_factory: optional factory if ModelToComponentFactory's default behavior needs to be tweaked.
+            normalize_manifest: Optional flag to indicate if the manifest should be normalized.
         """
         self.logger = logging.getLogger(f"airbyte.{self.name}")
-        # 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:
-            manifest["type"] = "DeclarativeSource"
-
+        self._should_normalize = normalize_manifest
+        self._declarative_component_schema = _get_declarative_component_schema()
         # If custom components are needed, locate and/or register them.
         self.components_module: ModuleType | None = get_registered_components_module(config=config)
+        # resolve all components in the manifest
+        self._source_config = self._preprocess_manifest(dict(source_config))
 
-        resolved_source_config = ManifestReferenceResolver().preprocess_manifest(manifest)
-        propagated_source_config = ManifestComponentTransformer().propagate_types_and_parameters(
-            "", resolved_source_config, {}
-        )
-        self._source_config = propagated_source_config
         self._debug = debug
         self._emit_connector_builder_messages = emit_connector_builder_messages
         self._constructor = (
@@ -105,22 +124,91 @@ def __init__(
         self._slice_logger: SliceLogger = (
             AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
         )
-
         self._config = config or {}
+
+        # validate resolved manifest against the declarative component schema
         self._validate_source()
 
+        # apply additional post-processing to the manifest
+        self._postprocess_manifest()
+
     @property
     def resolved_manifest(self) -> Mapping[str, Any]:
+        """
+        Returns the resolved manifest configuration for the source.
+
+        This property provides access to the internal source configuration as a mapping,
+        which contains all settings and parameters required to define the source's behavior.
+
+        Returns:
+            Mapping[str, Any]: The resolved source configuration manifest.
+        """
         return self._source_config
 
+    def _preprocess_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Preprocesses the provided manifest dictionary by resolving any manifest references.
+
+        This method modifies the input manifest in place, resolving references using the
+        ManifestReferenceResolver to ensure all references within the manifest are properly handled.
+
+        Args:
+            manifest (Dict[str, Any]): The manifest dictionary to preprocess and resolve references in.
+
+        Returns:
+            None
+        """
+        # 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 = self._fix_source_type(manifest)
+        # Resolve references in the manifest
+        resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest)
+        # Propagate types and parameters throughout the manifest
+        propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters(
+            "", resolved_manifest, {}
+        )
+
+        return propagated_manifest
+
+    def _postprocess_manifest(self) -> None:
+        """
+        Post-processes the manifest after validation.
+        This method is responsible for any additional modifications or transformations needed
+        after the manifest has been validated and before it is used in the source.
+        """
+        # apply manifest normalization, if required
+        self._normalize_manifest()
+
+    def _normalize_manifest(self) -> None:
+        """
+        This method is used to normalize the manifest. It should be called after the manifest has been validated.
+
+        Connector Builder UI rendering requires the manifest to be in a specific format.
+         - references have been resolved
+         - the commonly used definitions are extracted to the `definitions.linked.*`
+        """
+        if self._should_normalize:
+            normalizer = ManifestNormalizer(self._source_config, self._declarative_component_schema)
+            self._source_config = normalizer.normalize()
+
+    def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest.
+        """
+        if "type" not in manifest:
+            manifest["type"] = "DeclarativeSource"
+
+        return manifest
+
     @property
     def message_repository(self) -> MessageRepository:
         return self._message_repository
 
     @property
     def dynamic_streams(self) -> List[Dict[str, Any]]:
         return self._dynamic_stream_configs(
-            manifest=self._source_config, config=self._config, with_dynamic_stream_name=True
+            manifest=self._source_config,
+            config=self._config,
+            with_dynamic_stream_name=True,
         )
 
     @property
@@ -143,7 +231,10 @@ def connection_checker(self) -> ConnectionChecker:
 
     def streams(self, config: Mapping[str, Any]) -> List[Stream]:
         self._emit_manifest_debug_message(
-            extra_args={"source_name": self.name, "parsed_config": json.dumps(self._source_config)}
+            extra_args={
+                "source_name": self.name,
+                "parsed_config": json.dumps(self._source_config),
+            }
         )
 
         stream_configs = self._stream_configs(self._source_config) + self._dynamic_stream_configs(
@@ -156,9 +247,11 @@ def streams(self, config: Mapping[str, Any]) -> List[Stream]:
 
         source_streams = [
             self._constructor.create_component(
-                StateDelegatingStreamModel
-                if stream_config.get("type") == StateDelegatingStreamModel.__name__
-                else DeclarativeStreamModel,
+                (
+                    StateDelegatingStreamModel
+                    if stream_config.get("type") == StateDelegatingStreamModel.__name__
+                    else DeclarativeStreamModel
+                ),
                 stream_config,
                 config,
                 emit_connector_builder_messages=self._emit_connector_builder_messages,
@@ -174,7 +267,9 @@ def _initialize_cache_for_parent_streams(
     ) -> List[Dict[str, Any]]:
         parent_streams = set()
 
-        def update_with_cache_parent_configs(parent_configs: list[dict[str, Any]]) -> None:
+        def update_with_cache_parent_configs(
+            parent_configs: list[dict[str, Any]],
+        ) -> None:
             for parent_config in parent_configs:
                 parent_streams.add(parent_config["stream"]["name"])
                 if parent_config["stream"]["type"] == "StateDelegatingStream":
@@ -229,7 +324,10 @@ def spec(self, logger: logging.Logger) -> ConnectorSpecification:
         """
         self._configure_logger_level(logger)
         self._emit_manifest_debug_message(
-            extra_args={"source_name": self.name, "parsed_config": json.dumps(self._source_config)}
+            extra_args={
+                "source_name": self.name,
+                "parsed_config": json.dumps(self._source_config),
+            }
         )
 
         spec = self._source_config.get("spec")
@@ -266,25 +364,9 @@ 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}"
-            )
 
         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"
@@ -382,7 +464,9 @@ def _dynamic_stream_configs(
 
             # Create a resolver for dynamic components based on type
             components_resolver = self._constructor.create_component(
-                COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type], components_resolver_config, config
+                COMPONENTS_RESOLVER_TYPE_MAPPING[resolver_type],
+                components_resolver_config,
+                config,
             )
 
             stream_template_config = dynamic_definition["stream_template"]

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2209,7 +2209,7 @@ class HttpRequester(BaseModel):
     type: Literal["HttpRequester"]
     url_base: str = Field(
         ...,
-        description="Base URL of the API source. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.",
+        description="The Base URL of the API source. Do not put sensitive information (e.g. API tokens) into this field - Use the Authentication component for this.",
         examples=[
             "https://connect.squareup.com/v2",
             "{{ config['base_url'] or 'https://app.posthog.com'}}/api",
@@ -2220,7 +2220,7 @@ class HttpRequester(BaseModel):
     )
     path: Optional[str] = Field(
         None,
-        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.",
+        description="The 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.",
         examples=[
             "/products",
             "/quotes/{{ stream_partition['id'] }}/quote_line_groups",

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}")

airbyte_cdk/sources/declarative/parsers/manifest_component_transformer.py

@@ -4,7 +4,7 @@
 
 import copy
 import typing
-from typing import Any, Mapping, Optional
+from typing import Any, Dict, Mapping, Optional
 
 PARAMETERS_STR = "$parameters"
 
@@ -95,7 +95,7 @@ def propagate_types_and_parameters(
         declarative_component: Mapping[str, Any],
         parent_parameters: Mapping[str, Any],
         use_parent_parameters: Optional[bool] = None,
-    ) -> Mapping[str, Any]:
+    ) -> Dict[str, Any]:
         """
         Recursively transforms the specified declarative component and subcomponents to propagate parameters and insert the
         default component type if it was not already present. The resulting transformed components are a deep copy of the input