feat: Add unprivileged and config-free discover for declarative static schemas

airbyte_cdk/cli/source_declarative_manifest/_run.py

@@ -243,8 +243,8 @@ def _parse_inputs_into_config_catalog_state(
 ]:
     config = (
         ConcurrentDeclarativeSource.read_config(parsed_args.config)
-        if hasattr(parsed_args, "config")
-        else None
+        if hasattr(parsed_args, "config") and parsed_args.config
+        else {}
     )
     catalog = (
         ConcurrentDeclarativeSource.read_catalog(parsed_args.catalog)

airbyte_cdk/connector.py

@@ -31,8 +31,15 @@ def load_optional_package_file(package: str, filename: str) -> Optional[bytes]:
 
 
 class BaseConnector(ABC, Generic[TConfig]):
-    # configure whether the `check_config_against_spec_or_exit()` needs to be called
     check_config_against_spec: bool = True
+    """Configure whether `check_config_against_spec_or_exit()` needs to be called."""
+
+    check_config_during_discover: bool = True
+    """Determines whether config validation should be skipped during discovery.
+
+    By default, config validation is not skipped during discovery. This can be overridden
+    by sources that can provide catalog information without requiring authentication.
+    """
 
     @abstractmethod
     def configure(self, config: Mapping[str, Any], temp_dir: str) -> TConfig:

airbyte_cdk/entrypoint.py

@@ -105,7 +105,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"
         )
         discover_parser.add_argument(
             "--manifest-path",
@@ -177,19 +177,37 @@ 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_during_discover
+                ):
+                    # 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),
+                    )
+                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.")
                 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,
@@ -261,7 +279,7 @@ def discover(
         self, source_spec: ConnectorSpecification, config: TConfig
     ) -> Iterable[AirbyteMessage]:
         self.set_up_secret_filter(config, source_spec.connectionSpecification)
-        if self.source.check_config_against_spec:
+        if self.source.check_config_during_discover:
             self.validate_connection(source_spec, config)
         catalog = self.source.discover(self.logger, config)
 

airbyte_cdk/sources/declarative/manifest_declarative_source.py

@@ -66,6 +66,7 @@
 from airbyte_cdk.sources.declarative.resolvers import COMPONENTS_RESOLVER_TYPE_MAPPING
 from airbyte_cdk.sources.declarative.spec.spec import Spec
 from airbyte_cdk.sources.message import MessageRepository
+from airbyte_cdk.sources.source import Source
 from airbyte_cdk.sources.streams.core import Stream
 from airbyte_cdk.sources.types import Config, ConnectionDefinition
 from airbyte_cdk.sources.utils.slice_logger import (
@@ -97,6 +98,9 @@ def _get_declarative_component_schema() -> Dict[str, Any]:
 class ManifestDeclarativeSource(DeclarativeSource):
     """Declarative source defined by a manifest of low-code components that define source connector behavior"""
 
+    check_config_during_discover: bool = False
+    """Declarative sources default to not checking config before discovery."""
+
     def __init__(
         self,
         source_config: ConnectionDefinition,
@@ -148,6 +152,7 @@ def __init__(
         # apply additional post-processing to the manifest
         self._post_process_manifest()
 
+        self.check_config_during_discover = self._uses_dynamic_schema_loader()
         spec: Optional[Mapping[str, Any]] = self._source_config.get("spec")
         self._spec_component: Optional[Spec] = (
             self._constructor.create_component(SpecModel, spec, dict()) if spec else None
@@ -600,3 +605,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

airbyte_cdk/test/standard_tests/connector_base.py

@@ -8,12 +8,11 @@
 from pathlib import Path
 from typing import TYPE_CHECKING, cast
 
+import yaml
+from airbyte_protocol_dataclasses.models.airbyte_protocol import AirbyteConnectionStatus
 from boltons.typeutils import classproperty
 
-from airbyte_cdk.models import (
-    AirbyteMessage,
-    Type,
-)
+from airbyte_cdk.models import Status
 from airbyte_cdk.test import entrypoint_wrapper
 from airbyte_cdk.test.models import (
     ConnectorTestScenario,
@@ -122,6 +121,73 @@ def test_check(
             connector_root=self.get_connector_root_dir(),
         )
         assert len(result.connection_status_messages) == 1, (
-            f"Expected exactly one CONNECTION_STATUS message. "
-            "Got: {result.connection_status_messages!s}"
+            f"Expected exactly one CONNECTION_STATUS message. Got {len(result.connection_status_messages)}: \n"
+            + "\n".join([str(m) for m in result._messages or []])
+            + "\nErrors: "
+            + str(result.errors)
+            or "None"
+        )
+        conn_status = cast(
+            AirbyteConnectionStatus, result.connection_status_messages[0].connectionStatus
         )
+        if (
+            scenario.expected_outcome.expect_exception()
+            and conn_status.status == Status.SUCCEEDED
+            and not result.errors
+        ):
+            raise AssertionError(f"Expected error in `check` but got success.")
+    @classmethod
+    def get_connector_root_dir(cls) -> Path:
+        """Get the root directory of the connector."""
+        return find_connector_root([cls.get_test_class_dir(), Path.cwd()])
+
+    @classproperty
+    def acceptance_test_config_path(cls) -> Path:
+        """Get the path to the acceptance test config file."""
+        result = cls.get_connector_root_dir() / ACCEPTANCE_TEST_CONFIG
+        if result.exists():
+            return result
+
+        raise FileNotFoundError(f"Acceptance test config file not found at: {str(result)}")
+
+    @classmethod
+    def get_scenarios(
+        cls,
+    ) -> list[ConnectorTestScenario]:
+        """Get acceptance tests for a given category.
+
+        This has to be a separate function because pytest does not allow
+        parametrization of fixtures with arguments from the test class itself.
+        """
+        categories = ["connection", "spec"]
+        all_tests_config = yaml.safe_load(cls.acceptance_test_config_path.read_text())
+        if "acceptance_tests" not in all_tests_config:
+            raise ValueError(
+                f"Acceptance tests config not found in {cls.acceptance_test_config_path}."
+                f" Found only: {str(all_tests_config)}."
+            )
+
+        test_scenarios: list[ConnectorTestScenario] = []
+        for category in categories:
+            if (
+                category not in all_tests_config["acceptance_tests"]
+                or "tests" not in all_tests_config["acceptance_tests"][category]
+            ):
+                continue
+
+            test_scenarios.extend(
+                [
+                    ConnectorTestScenario.model_validate(test)
+                    for test in all_tests_config["acceptance_tests"][category]["tests"]
+                    if "config_path" in test and "iam_role" not in test["config_path"]
+                ]
+            )
+
+        connector_root = cls.get_connector_root_dir().absolute()
+        for test in test_scenarios:
+            if test.config_path:
+                test.config_path = connector_root / test.config_path
+            if test.configured_catalog_path:
+                test.configured_catalog_path = connector_root / test.configured_catalog_path
+
+        return test_scenarios

airbyte_cdk/test/standard_tests/source_base.py

@@ -6,6 +6,7 @@
 
 import pytest
 
+from airbyte_cdk.models import (
 from airbyte_cdk.models import (
     AirbyteMessage,
     AirbyteStream,
@@ -62,11 +63,9 @@ def test_discover(
     ) -> None:
         """Standard test for `discover`."""
         if scenario.expected_outcome.expect_exception():
-            # If the scenario expects an exception, we can't ensure it specifically would fail
-            # in discover, because some discover implementations do not need to make a connection.
-            # We skip this test in that case.
-            pytest.skip("Skipping discover test for scenario that expects an exception.")
-            return
+            pytest.skip(
+                "Skipping `discover` test because the scenario is expected to raise an exception."
+            )
 
         run_test_job(
             self.create_connector(scenario),
@@ -111,6 +110,15 @@ def test_basic_read(
         obtain the catalog of streams, and then it runs a `read` job to fetch
         records from those streams.
         """
+        check_result: entrypoint_wrapper.EntrypointOutput = run_test_job(
+            self.create_connector(scenario),
+            "check",
+            test_scenario=scenario,
+            connector_root=self.get_connector_root_dir(),
+        )
+        if scenario.expected_outcome.expect_exception() and check_result.errors:
+            # Expected failure and we got it. Return early.
+            return
         discover_result = run_test_job(
             self.create_connector(scenario),
             "discover",
@@ -138,7 +146,9 @@ def test_basic_read(
             connector_root=self.get_connector_root_dir(),
             catalog=configured_catalog,
         )
-
+        if scenario.expected_outcome.expect_exception() and not result.errors:
+            # By now we should have raised an exception.
+            raise AssertionError("Expected an error but got none.")
         if scenario.expected_outcome.expect_success() and not result.records:
             raise AssertionError("Expected records but got none.")
 
@@ -147,6 +157,14 @@ def test_fail_read_with_bad_catalog(
         scenario: ConnectorTestScenario,
     ) -> None:
         """Standard test for `read` when passed a bad catalog file."""
+        # Recreate the scenario with the same config but set the status to "failed".
+        scenario = ConnectorTestScenario(
+            config_dict=scenario.get_config_dict(
+                connector_root=scenario.connector_root,
+                empty_if_missing=False,
+            ),
+            status="failed",
+        )
         invalid_configured_catalog = ConfiguredAirbyteCatalog(
             streams=[
                 # Create ConfiguredAirbyteStream which is deliberately invalid