updated normalizer

Author: Oleksandr Bazarnov

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -1604,7 +1604,6 @@ 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.
@@ -1846,8 +1845,9 @@ definitions:
         type: string
         enum: [HttpRequester]
       url_base:
+        sharable: 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
@@ -1863,10 +1863,9 @@ 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.
+        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
@@ -1882,6 +1881,7 @@ definitions:
           - "/quotes/{{ stream_partition['id'] }}/quote_line_groups"
           - "/trades/{{ config['symbol_id'] }}/history"
       authenticator:
+        sharable: true
         title: Authenticator
         description: Authentication method to use for requests sent to the API.
         anyOf:
@@ -1895,7 +1895,6 @@ 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

@@ -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,
 )
@@ -89,7 +91,7 @@ def __init__(
         debug: bool = False,
         emit_connector_builder_messages: bool = False,
         component_factory: Optional[ModelToComponentFactory] = None,
-    ):
+    ) -> None:
         """
         Args:
             config: The provided config dict.
@@ -109,18 +111,24 @@ def __init__(
         # If custom components are needed, locate and/or register them.
         self.components_module: ModuleType | None = get_registered_components_module(config=config)
 
+        # resolve all `$ref` references in the manifest
         resolved_source_config = ManifestReferenceResolver().preprocess_manifest(manifest)
+        # resolve all components in the manifest
+        propagated_source_config = ManifestComponentTransformer().propagate_types_and_parameters(
+            "", resolved_source_config, {}
+        )
 
         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
+            # Connector Builder Ui rendering requires the manifest to be in a specific format.
+            # 1) references have been resolved
+            # 2) deprecated fields have been migrated
+            # 3) the commonly used definitions are extracted to the `definitions.shared.*`
+            # 4) ! the normalized manifest could be validated after the additional UI post-processing.
+            propagated_source_config = ManifestNormalizer(
+                propagated_source_config,
+                self._declarative_component_schema,
             ).normalize()
 
-        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
@@ -151,7 +159,9 @@ def message_repository(self) -> MessageRepository:
     @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
@@ -174,7 +184,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(
@@ -187,9 +200,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,
@@ -205,7 +220,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":
@@ -260,7 +277,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")
@@ -404,7 +424,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

@@ -49,7 +49,7 @@ class DynamicStreamCheckConfig(BaseModel):
     )
     stream_count: Optional[int] = Field(
         0,
-        description="Numbers of the streams to try reading from when running a check operation.",
+        description="The number of streams to attempt reading from during a check operation. If `stream_count` exceeds the total number of available streams, the minimum of the two values will be used.",
         title="Stream Count",
     )
 
@@ -2113,7 +2113,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",
@@ -2124,7 +2124,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/manifest_normalizer.py

@@ -84,9 +84,12 @@ def __init__(
     ) -> None:
         self._resolved_manifest = resolved_manifest
         self._declarative_schema = declarative_schema
+        self._normalized_manifest: ManifestType = copy.deepcopy(self._resolved_manifest)
         # get the tags marked as `sharable` in the component schema
         self._sharable_tags = _get_sharable_schema_tags(self._declarative_schema)
-        self._normalized_manifest: ManifestType = copy.deepcopy(self._resolved_manifest)
+
+    def to_json_str(self) -> str:
+        return json.dumps(self._normalized_manifest, indent=2)
 
     def normalize(self) -> ManifestType:
         """
@@ -193,7 +196,7 @@ def _replace_duplicates_with_refs(self, duplicates: DuplicatesType) -> None:
                 self._add_to_shared_definitions(type_key, key, value)
 
             # Replace occurrences with references
-            for path, parent_obj, value in occurrences:
+            for _, parent_obj, value in occurrences:
                 if is_shared_def:
                     if value == self._get_shared_definition_value(type_key, key):
                         parent_obj[key] = self._create_shared_definition_ref(type_key, key)
@@ -235,7 +238,7 @@ def _add_duplicate(
         """
 
         # create hash for each duplicate observed
-        value_to_hash = value if key is None else {key: value}
+        value_to_hash = {key: value} if key is not None else value
         duplicates[self._hash_object(value_to_hash)].append((current_path, obj, value))
 
     def _add_to_shared_definitions(

unit_tests/sources/declarative/interpolation/test_macros.py

@@ -30,7 +30,6 @@ def test_macros_export(test_name, fn_name, found_in_macros):
         assert fn_name not in macros
 
 
-@freeze_time("2022-01-01")
 @pytest.mark.parametrize(
     "input_value, format, input_format, expected_output",
     [

unit_tests/sources/declarative/parsers/test_manifest_normalizer.py

@@ -20,17 +20,29 @@ def test_deduplicate_manifest_when_multiple_url_base_are_resolved_and_most_frequ
     manifest_with_multiple_url_base,
     expected_manifest_with_multiple_url_base_normalized,
 ) -> None:
-    resolved_manifest = resolver.preprocess_manifest(manifest_with_multiple_url_base)
+    """
+    This test is to check that the manifest is normalized when multiple url_base are resolved
+    and the most frequent one is shared.
+    """
+
     schema = _get_declarative_component_schema()
+    resolved_manifest = resolver.preprocess_manifest(manifest_with_multiple_url_base)
     normalized_manifest = ManifestNormalizer(resolved_manifest, schema).normalize()
+
     assert normalized_manifest == expected_manifest_with_multiple_url_base_normalized
 
 
 def test_deduplicate_manifest_with_shared_definitions_url_base_are_present(
     manifest_with_url_base_shared_definition,
     expected_manifest_with_url_base_shared_definition_normalized,
 ) -> None:
-    resolved_manifest = resolver.preprocess_manifest(manifest_with_url_base_shared_definition)
+    """
+    This test is to check that the manifest is normalized when the `url_base` is shared
+    between the definitions and the `url_base` is present in the manifest.
+    """
+
     schema = _get_declarative_component_schema()
+    resolved_manifest = resolver.preprocess_manifest(manifest_with_url_base_shared_definition)
     normalized_manifest = ManifestNormalizer(resolved_manifest, schema).normalize()
+
     assert normalized_manifest == expected_manifest_with_url_base_shared_definition_normalized