feat: skip config validation during discovery for declarative sources that don't use DynamicSchemaLoader

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/entrypoint.py

@@ -93,7 +93,7 @@ def parse_args(args: List[str]) -> argparse.Namespace:
         )
         required_discover_parser = discover_parser.add_argument_group("required named arguments")
         required_discover_parser.add_argument(
-            "--config", type=str, required=True, help="path to the json configuration file"
+            "--config", type=str, required=False, help="path to the json configuration file"
         )
 
         # read
@@ -141,19 +141,34 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                 )
                 if cmd == "spec":
                     message = AirbyteMessage(type=Type.SPEC, spec=source_spec)
-                    yield from [
+                    yield from (
                         self.airbyte_message_to_string(queued_message)
                         for queued_message in self._emit_queued_messages(self.source)
-                    ]
+                    )
                     yield self.airbyte_message_to_string(message)
+                elif (
+                    cmd == "discover"
+                    and not parsed_args.config
+                    and not self.source.check_config_against_spec
+                ):
+                    # Connector supports unprivileged discover
+                    empty_config: dict[str, Any] = {}
+                    yield from (
+                        self.airbyte_message_to_string(queued_message)
+                        for queued_message in self._emit_queued_messages(self.source)
+                    )
+                    yield from map(
+                        AirbyteEntrypoint.airbyte_message_to_string,
+                        self.discover(source_spec, empty_config),
+                    )
                 else:
                     raw_config = self.source.read_config(parsed_args.config)
                     config = self.source.configure(raw_config, temp_dir)
 
-                    yield from [
+                    yield from (
                         self.airbyte_message_to_string(queued_message)
                         for queued_message in self._emit_queued_messages(self.source)
-                    ]
+                    )
                     if cmd == "check":
                         yield from map(
                             AirbyteEntrypoint.airbyte_message_to_string,

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,
@@ -109,6 +114,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]:
         return self._source_config
@@ -440,3 +447,35 @@ 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,164 @@
+#
+#
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from airbyte_cdk.models import AirbyteCatalog
+from airbyte_cdk.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
+from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit
+
+
+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)
+
+
+@patch(
+    "airbyte_cdk.sources.declarative.manifest_declarative_source.ManifestDeclarativeSource.streams"
+)
+def test_discover_with_dynamic_schema_loader_no_config(mock_streams):
+    """Test that discovery works without config when DynamicSchemaLoader is used."""
+    mock_stream = MagicMock()
+    mock_stream.name = "test_dynamic_stream"
+
+    mock_airbyte_stream = MagicMock()
+    type(mock_airbyte_stream).name = "test_dynamic_stream"
+    mock_stream.as_airbyte_stream.return_value = mock_airbyte_stream
+
+    mock_streams.return_value = [mock_stream]
+
+    source_config = {
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream"},
+        "streams": [
+            {
+                "name": "test_dynamic_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
+
+    logger = MagicMock()
+    catalog = source.discover(logger, {})
+
+    assert isinstance(catalog, AirbyteCatalog)
+    assert len(catalog.streams) == 1
+    assert catalog.streams[0].name == "test_dynamic_stream"
+
+
+@patch(
+    "airbyte_cdk.sources.declarative.manifest_declarative_source.ManifestDeclarativeSource.streams"
+)
+def test_discover_without_dynamic_schema_loader_no_config(mock_streams):
+    """Test that discovery validates config when DynamicSchemaLoader is not used."""
+    mock_stream = MagicMock()
+    mock_stream.name = "test_static_stream"
+
+    mock_airbyte_stream = MagicMock()
+    type(mock_airbyte_stream).name = "test_static_stream"
+    mock_stream.as_airbyte_stream.return_value = mock_airbyte_stream
+
+    mock_streams.return_value = [mock_stream]
+
+    source_config = {
+        "type": "DeclarativeSource",
+        "check": {"type": "CheckStream"},
+        "streams": [
+            {
+                "name": "test_static_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
+
+    logger = MagicMock()
+    catalog = source.discover(logger, {})
+
+    assert isinstance(catalog, AirbyteCatalog)
+    assert len(catalog.streams) == 1
+    assert catalog.streams[0].name == "test_static_stream"
+
+    assert source.check_config_against_spec is True

unit_tests/test_entrypoint.py

@@ -172,14 +172,13 @@ def test_parse_valid_args(
     ["cmd", "args"],
     [
         ("check", {"config": "config_path"}),
-        ("discover", {"config": "config_path"}),
         ("read", {"config": "config_path", "catalog": "catalog_path"}),
     ],
 )
 def test_parse_missing_required_args(
     cmd: str, args: MutableMapping[str, Any], entrypoint: AirbyteEntrypoint
 ):
-    required_args = {"check": ["config"], "discover": ["config"], "read": ["config", "catalog"]}
+    required_args = {"check": ["config"], "read": ["config", "catalog"]}
     for required_arg in required_args[cmd]:
         argcopy = deepcopy(args)
         del argcopy[required_arg]