Connector builder: handle empty catalog (#24184)

airbyte-cdk/python/connector_builder/main.py

@@ -4,7 +4,7 @@
 
 
 import sys
-from typing import Any, List, Mapping, Tuple
+from typing import Any, List, Mapping, Optional, Tuple
 
 from airbyte_cdk.connector import BaseConnector
 from airbyte_cdk.entrypoint import AirbyteEntrypoint
@@ -19,25 +19,36 @@ def create_source(config: Mapping[str, Any]) -> ManifestDeclarativeSource:
     return ManifestDeclarativeSource(manifest, True)
 
 
-def get_config_and_catalog_from_args(args: List[str]) -> Tuple[Mapping[str, Any], ConfiguredAirbyteCatalog]:
+def get_config_and_catalog_from_args(args: List[str]) -> Tuple[str, Mapping[str, Any], Optional[ConfiguredAirbyteCatalog]]:
     parsed_args = AirbyteEntrypoint.parse_args(args)
     config_path, catalog_path = parsed_args.config, parsed_args.catalog
     if parsed_args.command != "read":
         raise ValueError("Only read commands are allowed for Connector Builder requests.")
 
     config = BaseConnector.read_config(config_path)
-    catalog = ConfiguredAirbyteCatalog.parse_obj(BaseConnector.read_config(catalog_path))
+
+    if "__command" not in config:
+        raise ValueError(
+            f"Invalid config: `__command` should be provided at the root of the config but config only has keys {list(config.keys())}"
+        )
+
+    command = config["__command"]
+    if command == "test_read":
+        catalog = ConfiguredAirbyteCatalog.parse_obj(BaseConnector.read_config(catalog_path))
+    else:
+        catalog = None
 
     if "__injected_declarative_manifest" not in config:
         raise ValueError(
             f"Invalid config: `__injected_declarative_manifest` should be provided at the root of the config but config only has keys {list(config.keys())}"
         )
 
-    return config, catalog
+    return command, config, catalog
 
 
-def handle_connector_builder_request(source: ManifestDeclarativeSource, config: Mapping[str, Any], catalog: ConfiguredAirbyteCatalog):
-    command = config.get("__command")
+def handle_connector_builder_request(
+    source: ManifestDeclarativeSource, command: str, config: Mapping[str, Any], catalog: Optional[ConfiguredAirbyteCatalog]
+):
     if command == "resolve_manifest":
         return resolve_manifest(source)
     elif command == "test_read":
@@ -47,12 +58,9 @@ def handle_connector_builder_request(source: ManifestDeclarativeSource, config:
 
 
 def handle_request(args: List[str]):
-    config, catalog = get_config_and_catalog_from_args(args)
-    if "__command" in config:
-        source = create_source(config)
-        return handle_connector_builder_request(source, config, catalog).json(exclude_unset=True)
-    else:
-        raise ValueError("Missing __command argument in config file.")
+    command, config, catalog = get_config_and_catalog_from_args(args)
+    source = create_source(config)
+    return handle_connector_builder_request(source, command, config, catalog).json(exclude_unset=True)
 
 
 if __name__ == "__main__":

airbyte-cdk/python/unit_tests/connector_builder/test_connector_builder_handler.py

@@ -61,11 +61,7 @@ RESOLVE_MANIFEST_CONFIG = {
 TEST_READ_CONFIG = {
     "__injected_declarative_manifest": MANIFEST,
     "__command": "test_read",
-    "__test_read_config": {
-        "max_pages_per_slice": 2,
-        "max_slices": 5,
-        "max_records": 10
-    }
+    "__test_read_config": {"max_pages_per_slice": 2, "max_slices": 5, "max_records": 10},
 }
 
 DUMMY_CATALOG = {
@@ -73,16 +69,12 @@ DUMMY_CATALOG = {
         {
             "stream": {
                 "name": "dummy_stream",
-                "json_schema": {
-                    "$schema": "http://json-schema.org/draft-07/schema#",
-                    "type": "object",
-                    "properties": {}
-                },
+                "json_schema": {"$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": {}},
                 "supported_sync_modes": ["full_refresh"],
-                "source_defined_cursor": False
+                "source_defined_cursor": False,
             },
             "sync_mode": "full_refresh",
-            "destination_sync_mode": "overwrite"
+            "destination_sync_mode": "overwrite",
         }
     ]
 }
@@ -92,16 +84,12 @@ CONFIGURED_CATALOG = {
         {
             "stream": {
                 "name": _stream_name,
-                "json_schema": {
-                    "$schema": "http://json-schema.org/draft-07/schema#",
-                    "type": "object",
-                    "properties": {}
-                },
+                "json_schema": {"$schema": "http://json-schema.org/draft-07/schema#", "type": "object", "properties": {}},
                 "supported_sync_modes": ["full_refresh"],
-                "source_defined_cursor": False
+                "source_defined_cursor": False,
             },
             "sync_mode": "full_refresh",
-            "destination_sync_mode": "overwrite"
+            "destination_sync_mode": "overwrite",
         }
     ]
 }
@@ -158,9 +146,10 @@ def test_handle_test_read(valid_read_config_file, configured_catalog):
 
 def test_resolve_manifest(valid_resolve_manifest_config_file):
     config = copy.deepcopy(RESOLVE_MANIFEST_CONFIG)
-    config["__command"] = "resolve_manifest"
+    command = "resolve_manifest"
+    config["__command"] = command
     source = ManifestDeclarativeSource(MANIFEST)
-    resolved_manifest = handle_connector_builder_request(source, config, create_configured_catalog("dummy_stream"))
+    resolved_manifest = handle_connector_builder_request(source, command, config, create_configured_catalog("dummy_stream"))
 
     expected_resolved_manifest = {
         "type": "DeclarativeSource",
@@ -294,26 +283,38 @@ def test_read():
     source = ManifestDeclarativeSource(MANIFEST)
 
     real_record = AirbyteRecordMessage(data={"id": "1234", "key": "value"}, emitted_at=1, stream=_stream_name)
-    stream_read = StreamRead(logs=[{"message": "here be a log message"}],
-                             slices=[StreamReadSlicesInner(pages=[
-                                 StreamReadSlicesInnerPagesInner(records=[real_record], request=None, response=None)],
-                                 slice_descriptor=None, state=None)
-                             ],
-                             test_read_limit_reached=False, inferred_schema=None)
+    stream_read = StreamRead(
+        logs=[{"message": "here be a log message"}],
+        slices=[
+            StreamReadSlicesInner(
+                pages=[StreamReadSlicesInnerPagesInner(records=[real_record], request=None, response=None)],
+                slice_descriptor=None,
+                state=None,
+            )
+        ],
+        test_read_limit_reached=False,
+        inferred_schema=None,
+    )
 
-    expected_airbyte_message = AirbyteMessage(type=MessageType.RECORD,
-                                              record=AirbyteRecordMessage(stream=_stream_name, data={
-                                                  "logs": [{"message": "here be a log message"}],
-                                                  "slices": [{
-                                                      "pages": [{"records": [real_record], "request": None, "response": None}],
-                                                      "slice_descriptor": None,
-                                                      "state": None
-                                                  }],
-                                                  "test_read_limit_reached": False,
-                                                  "inferred_schema": None
-                                              }, emitted_at=1))
+    expected_airbyte_message = AirbyteMessage(
+        type=MessageType.RECORD,
+        record=AirbyteRecordMessage(
+            stream=_stream_name,
+            data={
+                "logs": [{"message": "here be a log message"}],
+                "slices": [
+                    {"pages": [{"records": [real_record], "request": None, "response": None}], "slice_descriptor": None, "state": None}
+                ],
+                "test_read_limit_reached": False,
+                "inferred_schema": None,
+            },
+            emitted_at=1,
+        ),
+    )
     with patch("connector_builder.message_grouper.MessageGrouper.get_message_groups", return_value=stream_read):
-        output_record = handle_connector_builder_request(source, config, ConfiguredAirbyteCatalog.parse_obj(CONFIGURED_CATALOG))
+        output_record = handle_connector_builder_request(
+            source, "test_read", config, ConfiguredAirbyteCatalog.parse_obj(CONFIGURED_CATALOG)
+        )
         output_record.record.emitted_at = 1
         assert output_record == expected_airbyte_message