feat: skip config validation during discovery for sources with DynamicSchemaLoader

Co-Authored-By: Aaron <AJ> Steers <[email protected]>

Author: devin-ai-integration[bot]

CHANGELOG.md

@@ -4,6 +4,10 @@ Newer updates can be found here: [GitHub Release Notes](https://github.com/airby
 
 # Changelog
 
+## Unreleased
+
+- Added automatic detection of DynamicSchemaLoader to skip config validation during discovery
+
 ## 6.5.2
 
 bugfix: Ensure that streams with partition router are not executed concurrently

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -58,7 +58,12 @@
 
 
 class ManifestDeclarativeSource(DeclarativeSource):
-    """Declarative source defined by a manifest of low-code components that define source connector behavior"""
+    """Declarative source defined by a manifest of low-code components that define source connector behavior
+    
+    If any stream in the source uses a DynamicSchemaLoader, config validation will be skipped during
+    discovery. This allows sources with dynamic schemas to run discovery without requiring authentication
+    when the schema endpoint doesn't need auth to provide catalog information.
+    """
 
     def __init__(
         self,
@@ -108,6 +113,8 @@ def __init__(
 
         self._config = config or {}
         self._validate_source()
+        
+        self.check_config_against_spec = not self._uses_dynamic_schema_loader()
 
     @property
     def resolved_manifest(self) -> Mapping[str, Any]:
@@ -440,3 +447,29 @@ def _dynamic_stream_configs(
 
     def _emit_manifest_debug_message(self, extra_args: dict[str, Any]) -> None:
         self.logger.debug("declarative source created from manifest", extra=extra_args)
+        
+    def _uses_dynamic_schema_loader(self) -> bool:
+        """
+        Determines if any stream in the source uses a DynamicSchemaLoader.
+        
+        DynamicSchemaLoader makes a separate call to retrieve schema information,
+        which might not require authentication, so we can skip config validation
+        during discovery when it's used.
+        
+        Returns:
+            bool: True if any stream uses a DynamicSchemaLoader, False otherwise.
+        """
+        for stream_config in self._stream_configs(self._source_config):
+            schema_loader = stream_config.get("schema_loader", {})
+            if isinstance(schema_loader, dict) and schema_loader.get("type") == "DynamicSchemaLoader":
+                return True
+                
+        dynamic_streams = self._source_config.get("dynamic_streams", [])
+        if dynamic_streams:
+            for dynamic_stream in dynamic_streams:
+                stream_template = dynamic_stream.get("stream_template", {})
+                schema_loader = stream_template.get("schema_loader", {})
+                if isinstance(schema_loader, dict) and schema_loader.get("type") == "DynamicSchemaLoader":
+                    return True
+                
+        return False

unit_tests/sources/declarative/test_manifest_declarative_source_dynamic_schema.py

@@ -0,0 +1,69 @@
+#
+#
+
+from unittest.mock import MagicMock
+
+import pytest
+
+from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
+
+
+def test_check_config_against_spec_with_dynamic_schema_loader():
+    """Test that check_config_against_spec is False when DynamicSchemaLoader is used."""
+    source_config = {
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream"},
+        "streams": [
+            {
+                "name": "test_stream",
+                "schema_loader": {
+                    "type": "DynamicSchemaLoader",
+                    "retriever": {
+                        "type": "SimpleRetriever",
+                        "requester": {"url_base": "https://example.com", "http_method": "GET"},
+                        "record_selector": {"extractor": {"field_path": []}},
+                    },
+                    "schema_type_identifier": {
+                        "key_pointer": ["name"],
+                    }
+                },
+                "retriever": {
+                    "type": "SimpleRetriever",
+                    "requester": {"url_base": "https://example.com", "http_method": "GET"},
+                    "record_selector": {"extractor": {"field_path": []}},
+                }
+            }
+        ],
+        "version": "0.1.0"
+    }
+    
+    source = ManifestDeclarativeSource(source_config=source_config)
+    
+    assert source.check_config_against_spec is False
+
+
+def test_check_config_against_spec_without_dynamic_schema_loader():
+    """Test that check_config_against_spec is True when DynamicSchemaLoader is not used."""
+    source_config = {
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream"},
+        "streams": [
+            {
+                "name": "test_stream",
+                "schema_loader": {
+                    "type": "InlineSchemaLoader",
+                    "schema": {}
+                },
+                "retriever": {
+                    "type": "SimpleRetriever",
+                    "requester": {"url_base": "https://example.com", "http_method": "GET"},
+                    "record_selector": {"extractor": {"field_path": []}},
+                }
+            }
+        ],
+        "version": "0.1.0"
+    }
+    
+    source = ManifestDeclarativeSource(source_config=source_config)
+    
+    assert source.check_config_against_spec is True