feat: Add unprivileged discover support for declarative sources with static schemas

- Move check_config_during_discover logic to ConcurrentDeclarativeSource (the non-deprecated class)
- Add _uses_dynamic_schema_loader() method to detect DynamicSchemaLoader usage
- Set check_config_during_discover=False for sources without DynamicSchemaLoader
- Fix docstring in ManifestDeclarativeSource to correctly describe DynamicSchemaLoader behavior
- Add entrypoint unit tests for discover with/without config
- Improve error message when config is required but not provided
- Update test imports to use legacy module location

This enables declarative sources with static schemas (InlineSchemaLoader) to run
discover without credentials, while sources with DynamicSchemaLoader still require
config for authentication.

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

Author: devin-ai-integration[bot]

airbyte_cdk/entrypoint.py

@@ -199,7 +199,11 @@ def run(self, parsed_args: argparse.Namespace) -> Iterable[str]:
                     )
                 elif parsed_args.config is None:
                     # Raise a helpful error message if we reach here with no config.
-                    raise ValueError("The '--config' arg is required but was not provided.")
+                    raise ValueError(
+                        "The '--config' argument is required but was not provided. "
+                        "This connector does not support unprivileged discovery. "
+                        "Please provide a valid configuration file using the --config flag."
+                    )
                 else:
                     raw_config = self.source.read_config(parsed_args.config)
                     config = self.source.configure(raw_config, temp_dir)

airbyte_cdk/legacy/sources/declarative/manifest_declarative_source.py

@@ -624,11 +624,12 @@ 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.
+        which might require authentication. When present, config validation cannot
+        be skipped during discovery.
 
         Returns:
-            bool: True if any stream uses a DynamicSchemaLoader, False otherwise.
+            bool: True if any stream uses a DynamicSchemaLoader (config required for discover),
+                  False otherwise (unprivileged discover may be supported).
         """
         empty_config: Dict[str, Any] = {}
         for stream_config in self._stream_configs(self._source_config, empty_config):

airbyte_cdk/sources/declarative/concurrent_declarative_source.py

@@ -242,6 +242,8 @@ def __init__(
             message_repository=self._message_repository,
         )
 
+        self.check_config_during_discover = self._uses_dynamic_schema_loader()
+
     def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]:
         """
         Preprocesses the provided manifest dictionary by resolving any manifest references.
@@ -651,3 +653,36 @@ def _select_streams(
                     as_airbyte_message(configured_stream.stream, AirbyteStreamStatus.INCOMPLETE)
                 )
         return abstract_streams
+
+    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 require authentication. When present, config validation cannot
+        be skipped during discovery.
+
+        Returns:
+            bool: True if any stream uses a DynamicSchemaLoader (config required for discover),
+                  False otherwise (unprivileged discover may be supported).
+        """
+        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_stream_definitions = self._source_config.get("dynamic_streams", [])
+        if dynamic_stream_definitions:
+            for dynamic_definition in dynamic_stream_definitions:
+                stream_template = dynamic_definition.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

@@ -8,8 +8,10 @@
 
 import pytest
 
+from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import (
+    ManifestDeclarativeSource,
+)
 from airbyte_cdk.models import AirbyteCatalog
-from airbyte_cdk.legacy.sources.declarative.manifest_declarative_source import ManifestDeclarativeSource
 from airbyte_cdk.sources.utils.schema_helpers import check_config_against_spec_or_exit
 
 

unit_tests/test_entrypoint.py

@@ -855,3 +855,77 @@ def test_given_serialization_error_using_orjson_then_fallback_on_json(
     # There will be multiple messages here because the fixture `entrypoint` sets a control message. We only care about records here
     record_messages = list(filter(lambda message: "RECORD" in message, messages))
     assert len(record_messages) == 2
+
+def test_run_discover_without_config_when_supported(mocker):
+    """Test that discover works without config when check_config_during_discover is False."""
+    # Create a mock source that supports unprivileged discover
+    mock_source = MockSource()
+    mock_source.check_config_during_discover = False
+    
+    message_repository = MagicMock()
+    message_repository.consume_queue.return_value = []
+    mocker.patch.object(
+        MockSource,
+        "message_repository",
+        new_callable=mocker.PropertyMock,
+        return_value=message_repository,
+    )
+    
+    entrypoint = AirbyteEntrypoint(mock_source)
+    
+    parsed_args = Namespace(command="discover", config=None)
+    expected_catalog = AirbyteCatalog(
+        streams=[
+            AirbyteStream(
+                name="test_stream", 
+                json_schema={"type": "object"}, 
+                supported_sync_modes=[SyncMode.full_refresh]
+            )
+        ]
+    )
+    
+    spec = ConnectorSpecification(connectionSpecification={})
+    mocker.patch.object(MockSource, "spec", return_value=spec)
+    mocker.patch.object(MockSource, "discover", return_value=expected_catalog)
+    
+    messages = list(entrypoint.run(parsed_args))
+    
+    # Should successfully return catalog without config
+    assert len(messages) == 1
+    assert _wrap_message(expected_catalog) == messages[0]
+    
+    # Verify discover was called with empty config
+    MockSource.discover.assert_called_once()
+    call_args = MockSource.discover.call_args
+    assert call_args[0][1] == {}  # config argument should be empty dict
+
+
+def test_run_discover_without_config_when_not_supported(mocker):
+    """Test that discover fails with helpful error when config is required but not provided."""
+    # Create a mock source that requires config for discover
+    mock_source = MockSource()
+    mock_source.check_config_during_discover = True
+    
+    message_repository = MagicMock()
+    message_repository.consume_queue.return_value = []
+    mocker.patch.object(
+        MockSource,
+        "message_repository",
+        new_callable=mocker.PropertyMock,
+        return_value=message_repository,
+    )
+    
+    entrypoint = AirbyteEntrypoint(mock_source)
+    
+    parsed_args = Namespace(command="discover", config=None)
+    
+    spec = ConnectorSpecification(connectionSpecification={})
+    mocker.patch.object(MockSource, "spec", return_value=spec)
+    
+    # Should raise ValueError with helpful message
+    with pytest.raises(ValueError) as exc_info:
+        list(entrypoint.run(parsed_args))
+    
+    error_message = str(exc_info.value)
+    assert "The '--config' argument is required but was not provided" in error_message
+    assert "does not support unprivileged discovery" in error_message