Add full-resolve-manifest command to connector builder

Author: lazebnyi

airbyte_cdk/connector_builder/connector_builder_handler.py

@@ -43,7 +43,7 @@ class TestReadLimits:
 def get_limits(config: Mapping[str, Any]) -> TestReadLimits:
     command_config = config.get("__test_read_config", {})
     max_pages_per_slice = (
-        command_config.get(MAX_PAGES_PER_SLICE_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE
+            command_config.get(MAX_PAGES_PER_SLICE_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_PAGES_PER_SLICE
     )
     max_slices = command_config.get(MAX_SLICES_KEY) or DEFAULT_MAXIMUM_NUMBER_OF_SLICES
     max_records = command_config.get(MAX_RECORDS_KEY) or DEFAULT_MAXIMUM_RECORDS
@@ -67,11 +67,11 @@ def create_source(config: Mapping[str, Any], limits: TestReadLimits) -> Manifest
 
 
 def read_stream(
-    source: DeclarativeSource,
-    config: Mapping[str, Any],
-    configured_catalog: ConfiguredAirbyteCatalog,
-    state: List[AirbyteStateMessage],
-    limits: TestReadLimits,
+        source: DeclarativeSource,
+        config: Mapping[str, Any],
+        configured_catalog: ConfiguredAirbyteCatalog,
+        state: List[AirbyteStateMessage],
+        limits: TestReadLimits,
 ) -> AirbyteMessage:
     try:
         test_read_handler = TestReader(
@@ -117,5 +117,28 @@ 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)
+
+        return AirbyteMessage(
+            type=Type.RECORD,
+            record=AirbyteRecordMessage(
+                data={"manifest": manifest},
+                emitted_at=_emitted_at(),
+                stream="full_resolve_manifest",
+            ),
+        )
+    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

@@ -15,6 +15,7 @@
     get_limits,
     read_stream,
     resolve_manifest,
+    full_resolve_manifest,
 )
 from airbyte_cdk.entrypoint import AirbyteEntrypoint
 from airbyte_cdk.models import (
@@ -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/manifest_declarative_source.py

@@ -106,6 +106,7 @@ def __init__(
             AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger()
         )
 
+        self._config = config
         self._validate_source()
 
     @property
@@ -116,6 +117,10 @@ 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)
+
     @property
     def connection_checker(self) -> ConnectionChecker:
         check = self._source_config["check"]
@@ -343,7 +348,6 @@ def _stream_configs(self, manifest: Mapping[str, Any]) -> List[Dict[str, Any]]:
         # This has a warning flag for static, but after we finish part 4 we'll replace manifest with self._source_config
         stream_configs: List[Dict[str, Any]] = manifest.get("streams", [])
         for s in stream_configs:
-            s["dynamic_stream_name"] = None
             if "type" not in s:
                 s["type"] = "DeclarativeStream"
         return stream_configs