fix: add basic error handling to manifest-server calls

Author: ChristoGrab

airbyte_cdk/manifest_server/routers/manifest.py

@@ -5,8 +5,10 @@
 import jsonschema
 from airbyte_protocol_dataclasses.models import AirbyteStateMessage, ConfiguredAirbyteCatalog
 from fastapi import APIRouter, Depends, HTTPException
+from fastapi.responses import JSONResponse
 
 from airbyte_cdk.models import AirbyteStateMessageSerializer
+from airbyte_cdk.utils.traced_exception import AirbyteTracedException
 from airbyte_cdk.sources.declarative.concurrent_declarative_source import (
     ConcurrentDeclarativeSource,
 )
@@ -98,15 +100,21 @@ def test_read(request: StreamTestReadRequest) -> StreamReadResponse:
     )
 
     runner = ManifestCommandProcessor(source)
-    cdk_result = runner.test_read(
-        config_dict,
-        catalog,
-        converted_state,
-        request.record_limit,
-        request.page_limit,
-        request.slice_limit,
-    )
-    return StreamReadResponse.model_validate(asdict(cdk_result))
+    try:
+        cdk_result = runner.test_read(
+            config_dict,
+            catalog,
+            converted_state,
+            request.record_limit,
+            request.page_limit,
+            request.slice_limit,
+        )
+        return StreamReadResponse.model_validate(asdict(cdk_result))
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error reading stream: {str(exc)}"
+        )
+        raise HTTPException(status_code=400, detail=error.message)
 
 
 @router.post("/check", operation_id="check")
@@ -119,10 +127,16 @@ def check(request: CheckRequest) -> CheckResponse:
             project_id=request.context.project_id,
         )
 
-    source = safe_build_source(request.manifest.model_dump(), request.config.model_dump())
-    runner = ManifestCommandProcessor(source)
-    success, message = runner.check_connection(request.config.model_dump())
-    return CheckResponse(success=success, message=message)
+    try:
+        source = safe_build_source(request.manifest.model_dump(), request.config.model_dump())
+        runner = ManifestCommandProcessor(source)
+        success, message = runner.check_connection(request.config.model_dump())
+        return CheckResponse(success=success, message=message)
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error checking connection: {str(exc)}"
+        )
+        raise HTTPException(status_code=400, detail=error.message)
 
 
 @router.post("/discover", operation_id="discover")
@@ -135,12 +149,21 @@ def discover(request: DiscoverRequest) -> DiscoverResponse:
             project_id=request.context.project_id,
         )
 
-    source = safe_build_source(request.manifest.model_dump(), request.config.model_dump())
-    runner = ManifestCommandProcessor(source)
-    catalog = runner.discover(request.config.model_dump())
-    if catalog is None:
-        raise HTTPException(status_code=422, detail="Connector did not return a discovered catalog")
-    return DiscoverResponse(catalog=catalog)
+    try:
+        source = safe_build_source(request.manifest.model_dump(), request.config.model_dump())
+        runner = ManifestCommandProcessor(source)
+        catalog = runner.discover(request.config.model_dump())
+        if catalog is None:
+            raise HTTPException(status_code=422, detail="Connector did not return a discovered catalog")
+        return DiscoverResponse(catalog=catalog)
+    except HTTPException:
+        # Re-raise HTTPExceptions as-is (like the catalog None check above)
+        raise
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error discovering streams: {str(exc)}"
+        )
+        raise HTTPException(status_code=400, detail=error.message)
 
 
 @router.post("/resolve", operation_id="resolve")
@@ -153,8 +176,14 @@ def resolve(request: ResolveRequest) -> ManifestResponse:
             project_id=request.context.project_id,
         )
 
-    source = safe_build_source(request.manifest.model_dump(), {})
-    return ManifestResponse(manifest=Manifest(**source.resolved_manifest))
+    try:
+        source = safe_build_source(request.manifest.model_dump(), {})
+        return ManifestResponse(manifest=Manifest(**source.resolved_manifest))
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error resolving manifest: {str(exc)}"
+        )
+        raise HTTPException(status_code=400, detail=error.message)
 
 
 @router.post("/full_resolve", operation_id="fullResolve")
@@ -171,21 +200,27 @@ def full_resolve(request: FullResolveRequest) -> ManifestResponse:
             project_id=request.context.project_id,
         )
 
-    source = safe_build_source(request.manifest.model_dump(), request.config.model_dump())
-    manifest = {**source.resolved_manifest}
-    streams = manifest.get("streams", [])
-    for stream in streams:
-        stream["dynamic_stream_name"] = None
-
-    mapped_streams: Dict[str, List[Dict[str, Any]]] = {}
-    for stream in source.dynamic_streams:
-        generated_streams = mapped_streams.setdefault(stream["dynamic_stream_name"], [])
-
-        if len(generated_streams) < request.stream_limit:
-            generated_streams += [stream]
-
-    for generated_streams_list in mapped_streams.values():
-        streams.extend(generated_streams_list)
-
-    manifest["streams"] = streams
-    return ManifestResponse(manifest=Manifest(**manifest))
+    try:
+        source = safe_build_source(request.manifest.model_dump(), request.config.model_dump())
+        manifest = {**source.resolved_manifest}
+        streams = manifest.get("streams", [])
+        for stream in streams:
+            stream["dynamic_stream_name"] = None
+
+        mapped_streams: Dict[str, List[Dict[str, Any]]] = {}
+        for stream in source.dynamic_streams:
+            generated_streams = mapped_streams.setdefault(stream["dynamic_stream_name"], [])
+
+            if len(generated_streams) < request.stream_limit:
+                generated_streams += [stream]
+
+        for generated_streams_list in mapped_streams.values():
+            streams.extend(generated_streams_list)
+
+        manifest["streams"] = streams
+        return ManifestResponse(manifest=Manifest(**manifest))
+    except Exception as exc:
+        error = AirbyteTracedException.from_exception(
+            exc, message=f"Error full resolving manifest: {str(exc)}"
+        )
+        raise HTTPException(status_code=400, detail=error.message)

unit_tests/manifest_server/routers/test_manifest.py

@@ -527,3 +527,126 @@ def test_discover_endpoint_missing_catalog(
         assert response.status_code == 422
         data = response.json()
         assert "Connector did not return a discovered catalog" in data["detail"]
+
+    # Test cases for error handling improvements
+    @patch("airbyte_cdk.manifest_server.routers.manifest.ManifestCommandProcessor")
+    @patch("airbyte_cdk.manifest_server.routers.manifest.build_source")
+    def test_test_read_cdk_error_handling(
+        self, mock_build_source, mock_runner_class, sample_manifest, sample_config, mock_source
+    ):
+        """Test that CDK errors in test_read are properly caught and converted to HTTP 400."""
+        request_data = {
+            "manifest": sample_manifest,
+            "config": sample_config,
+            "stream_name": "products"
+        }
+
+        mock_build_source.return_value = mock_source
+
+        mock_runner = Mock()
+        # Simulate a CDK error (like datetime parsing error)
+        mock_runner.test_read.side_effect = ValueError("time data '' does not match format '%Y-%m-%dT%H:%M:%SZ'")
+        mock_runner_class.return_value = mock_runner
+
+        response = client.post("/v1/manifest/test_read", json=request_data)
+
+        assert response.status_code == 400
+        data = response.json()
+        assert "detail" in data
+        assert "Error reading stream:" in data["detail"]
+        assert "time data" in data["detail"]
+        assert "does not match format" in data["detail"]
+
+    @patch("airbyte_cdk.manifest_server.routers.manifest.ManifestCommandProcessor")
+    @patch("airbyte_cdk.manifest_server.routers.manifest.build_source")
+    def test_check_cdk_error_handling(
+        self, mock_build_source, mock_runner_class, sample_manifest, sample_config, mock_source
+    ):
+        """Test that CDK errors in check are properly caught and converted to HTTP 400."""
+        request_data = {
+            "manifest": sample_manifest,
+            "config": sample_config,
+        }
+
+        mock_build_source.return_value = mock_source
+
+        mock_runner = Mock()
+        # Simulate a CDK error (like connection error)
+        mock_runner.check_connection.side_effect = ConnectionError("Failed to connect to API")
+        mock_runner_class.return_value = mock_runner
+
+        response = client.post("/v1/manifest/check", json=request_data)
+
+        assert response.status_code == 400
+        data = response.json()
+        assert "detail" in data
+        assert "Error checking connection:" in data["detail"]
+        assert "Failed to connect to API" in data["detail"]
+
+    @patch("airbyte_cdk.manifest_server.routers.manifest.ManifestCommandProcessor")
+    @patch("airbyte_cdk.manifest_server.routers.manifest.build_source")
+    def test_discover_cdk_error_handling(
+        self, mock_build_source, mock_runner_class, sample_manifest, sample_config, mock_source
+    ):
+        """Test that CDK errors in discover are properly caught and converted to HTTP 400."""
+        request_data = {
+            "manifest": sample_manifest,
+            "config": sample_config,
+        }
+
+        mock_build_source.return_value = mock_source
+
+        mock_runner = Mock()
+        # Simulate a CDK error
+        mock_runner.discover.side_effect = RuntimeError("Schema validation failed")
+        mock_runner_class.return_value = mock_runner
+
+        response = client.post("/v1/manifest/discover", json=request_data)
+
+        assert response.status_code == 400
+        data = response.json()
+        assert "detail" in data
+        assert "Error discovering streams:" in data["detail"]
+        assert "Schema validation failed" in data["detail"]
+
+    @patch("airbyte_cdk.manifest_server.routers.manifest.build_source")
+    def test_resolve_cdk_error_handling(
+        self, mock_build_source, sample_manifest
+    ):
+        """Test that CDK errors in resolve are properly caught and converted to HTTP 400."""
+        request_data = {
+            "manifest": sample_manifest,
+        }
+
+        # Simulate a CDK error during source building
+        mock_build_source.side_effect = AttributeError("'NoneType' object has no attribute 'get'")
+
+        response = client.post("/v1/manifest/resolve", json=request_data)
+
+        assert response.status_code == 400
+        data = response.json()
+        assert "detail" in data
+        assert "Error resolving manifest:" in data["detail"]
+        assert "'NoneType' object has no attribute 'get'" in data["detail"]
+
+    @patch("airbyte_cdk.manifest_server.routers.manifest.build_source")
+    def test_full_resolve_cdk_error_handling(
+        self, mock_build_source, sample_manifest, sample_config
+    ):
+        """Test that CDK errors in full_resolve are properly caught and converted to HTTP 400."""
+        request_data = {
+            "manifest": sample_manifest,
+            "config": sample_config,
+            "stream_limit": 10
+        }
+
+        # Simulate a CDK error during source building
+        mock_build_source.side_effect = KeyError("Missing required field 'streams'")
+
+        response = client.post("/v1/manifest/full_resolve", json=request_data)
+
+        assert response.status_code == 400
+        data = response.json()
+        assert "detail" in data
+        assert "Error full resolving manifest:" in data["detail"]
+        assert "Missing required field 'streams'" in data["detail"]