feat(connector-builder): add full_resolve command to connector builder (#442)

Co-authored-by: octavia-squidington-iii <[email protected]>

Author: lazebnyi

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -117,5 +117,30 @@ def resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
         return error.as_airbyte_message()
 
 
+def full_resolve_manifest(source: ManifestDeclarativeSource) -> AirbyteMessage:
+    try:
+        manifest = {**source.resolved_manifest}
+        streams = manifest.get("streams", [])
+        for stream in streams:
+            stream["dynamic_stream_name"] = None
+        streams.extend(source.dynamic_streams)
+        manifest["streams"] = streams
+        return AirbyteMessage(
+            type=Type.RECORD,
+            record=AirbyteRecordMessage(
+                data={"manifest": manifest},
+                emitted_at=_emitted_at(),
+                stream="full_resolve_manifest",
+            ),
+        )
+    except AirbyteTracedException as exc:
+        return exc.as_airbyte_message()
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error full resolving manifest: {str(exc)}"
+        )
+        return error.as_airbyte_message()
+
+
 def _emitted_at() -> int:
     return ab_datetime_now().to_epoch_millis()

airbyte_cdk/connector_builder/main.py

@@ -12,6 +12,7 @@
 from airbyte_cdk.connector_builder.connector_builder_handler import (
     TestReadLimits,
     create_source,
+    full_resolve_manifest,
     get_limits,
     read_stream,
     resolve_manifest,
@@ -81,6 +82,8 @@ def handle_connector_builder_request(
             catalog is not None
         ), "`test_read` requires a valid `ConfiguredAirbyteCatalog`, got None."
         return read_stream(source, config, catalog, state, limits)
+    elif command == "full_resolve_manifest":
+        return full_resolve_manifest(source)
     else:
         raise ValueError(f"Unrecognized command {command}.")
 

airbyte_cdk/sources/declarative/declarative_component_schema.yaml

@@ -3766,6 +3766,13 @@ definitions:
       type:
         type: string
         enum: [DynamicDeclarativeStream]
+      name:
+        title: Name
+        description: The dynamic stream name.
+        type: string
+        default: ""
+        example:
+          - "Tables"
       stream_template:
         title: Stream Template
         description: Reference to the stream template.

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -106,6 +106,7 @@ def __init__(
             AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
         )
 
+        self._config = config or {}
         self._validate_source()
 
     @property
@@ -116,6 +117,12 @@ def resolved_manifest(self) -> Mapping[str, Any]:
     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
+        )
+
     @property
     def connection_checker(self) -> ConnectionChecker:
         check = self._source_config["check"]
@@ -348,13 +355,16 @@ def _stream_configs(self, manifest: Mapping[str, Any]) -> List[Dict[str, Any]]:
         return stream_configs
 
     def _dynamic_stream_configs(
-        self, manifest: Mapping[str, Any], config: Mapping[str, Any]
+        self,
+        manifest: Mapping[str, Any],
+        config: Mapping[str, Any],
+        with_dynamic_stream_name: Optional[bool] = None,
     ) -> List[Dict[str, Any]]:
         dynamic_stream_definitions: List[Dict[str, Any]] = manifest.get("dynamic_streams", [])
         dynamic_stream_configs: List[Dict[str, Any]] = []
         seen_dynamic_streams: Set[str] = set()
 
-        for dynamic_definition in dynamic_stream_definitions:
+        for dynamic_definition_index, dynamic_definition in enumerate(dynamic_stream_definitions):
             components_resolver_config = dynamic_definition["components_resolver"]
 
             if not components_resolver_config:
@@ -393,6 +403,11 @@ def _dynamic_stream_configs(
                 # Ensure that each stream is created with a unique name
                 name = dynamic_stream.get("name")
 
+                if with_dynamic_stream_name:
+                    dynamic_stream["dynamic_stream_name"] = dynamic_definition.get(
+                        "name", f"dynamic_stream_{dynamic_definition_index}"
+                    )
+
                 if not isinstance(name, str):
                     raise ValueError(
                         f"Expected stream name {name} to be a string, got {type(name)}."

airbyte_cdk/sources/declarative/models/declarative_component_schema.py

@@ -2489,6 +2489,9 @@ class HttpComponentsResolver(BaseModel):
 
 class DynamicDeclarativeStream(BaseModel):
     type: Literal["DynamicDeclarativeStream"]
+    name: Optional[str] = Field(
+        "", description="The dynamic stream name.", example=["Tables"], title="Name"
+    )
     stream_template: DeclarativeStream = Field(
         ..., description="Reference to the stream template.", title="Stream Template"
     )