diff --git a/airbyte_cdk/cli/source_declarative_manifest/_run.py b/airbyte_cdk/cli/source_declarative_manifest/_run.py index 1b64e51252..b6692f9e18 100644 --- a/airbyte_cdk/cli/source_declarative_manifest/_run.py +++ b/airbyte_cdk/cli/source_declarative_manifest/_run.py @@ -37,6 +37,7 @@ AirbyteTraceMessage, ConfiguredAirbyteCatalog, ConnectorSpecificationSerializer, + FailureType, TraceType, Type, ) @@ -108,6 +109,9 @@ def _get_local_yaml_source(args: list[str]) -> SourceLocalYaml: config_path=parsed_args.config if hasattr(parsed_args, "config") else None, ) except Exception as error: + is_config_free_discover = args[:1] == ["discover"] and not any( + arg.startswith("--config") for arg in args + ) print( orjson.dumps( AirbyteMessageSerializer.dump( @@ -117,8 +121,15 @@ def _get_local_yaml_source(args: list[str]) -> SourceLocalYaml: type=TraceType.ERROR, emitted_at=ab_datetime_now().to_epoch_millis(), error=AirbyteErrorTraceMessage( - message=f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}", + message=( + f"This connector could not run `discover` without a config. Provide one with `--config`. Error: {error}" + if is_config_free_discover + else f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}" + ), stack_trace=traceback.format_exc(), + failure_type=FailureType.config_error + if is_config_free_discover + else None, ), ), ) @@ -222,6 +233,9 @@ def create_declarative_source( custom_components_trusted=False, ) except Exception as error: + is_config_free_discover = args[:1] == ["discover"] and not any( + arg.startswith("--config") for arg in args + ) print( orjson.dumps( AirbyteMessageSerializer.dump( @@ -231,8 +245,15 @@ def create_declarative_source( type=TraceType.ERROR, emitted_at=ab_datetime_now().to_epoch_millis(), error=AirbyteErrorTraceMessage( - message=f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}", + message=( + f"This connector could not run `discover` without a config. Provide one with `--config`. Error: {error}" + if is_config_free_discover + else f"Error starting the sync. This could be due to an invalid configuration or catalog. Please contact Support for assistance. Error: {error}" + ), stack_trace=traceback.format_exc(), + failure_type=FailureType.config_error + if is_config_free_discover + else None, ), ), ) @@ -251,8 +272,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) diff --git a/airbyte_cdk/connector.py b/airbyte_cdk/connector.py index a7c13dcda9..aaa9b68726 100644 --- a/airbyte_cdk/connector.py +++ b/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 + """Whether `discover` requires a config and validates it against the spec. + + Set to `False` for sources that can build their catalog without a config (and therefore + without credentials), which allows `discover` to run without `--config`. + """ @abstractmethod def configure(self, config: Mapping[str, Any], temp_dir: str) -> TConfig: diff --git a/airbyte_cdk/entrypoint.py b/airbyte_cdk/entrypoint.py index c2432237bc..18776bab74 100644 --- a/airbyte_cdk/entrypoint.py +++ b/airbyte_cdk/entrypoint.py @@ -105,9 +105,11 @@ def parse_args(args: List[str]) -> argparse.Namespace: help="outputs a catalog describing the source's schema", parents=[parent_parser], ) - 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" + discover_parser.add_argument( + "--config", + type=str, + required=False, + help="path to the json configuration file (optional if the source supports config-free discover)", ) discover_parser.add_argument( "--manifest-path", @@ -179,19 +181,38 @@ 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 + ): + 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_without_config(), + ) + elif cmd == "discover" and not parsed_args.config: + raise AirbyteTracedException( + message="This connector requires a config to run `discover`. Provide one with `--config`.", + internal_message="`--config` was not provided and the source does not support config-free discover.", + failure_type=FailureType.config_error, + ) 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, @@ -267,6 +288,23 @@ def discover( yield from self._emit_queued_messages(self.source) yield AirbyteMessage(type=Type.CATALOG, catalog=catalog) + def _discover_without_config(self) -> Iterable[AirbyteMessage]: + """Run `discover` with an empty config, for sources that support config-free discover.""" + try: + catalog = self.source.discover(self.logger, {}) + except AirbyteTracedException: + raise + except Exception as error: + raise AirbyteTracedException( + message="This connector could not run `discover` without a config. Provide one with `--config`.", + internal_message=f"Config-free discover failed: {error!r}", + failure_type=FailureType.config_error, + exception=error, + ) from error + + yield from self._emit_queued_messages(self.source) + yield AirbyteMessage(type=Type.CATALOG, catalog=catalog) + def read( self, source_spec: ConnectorSpecification, config: TConfig, catalog: Any, state: list[Any] ) -> Iterable[AirbyteMessage]: diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 055576f45f..67240c69dd 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -271,6 +271,8 @@ def __init__( message_repository=self._message_repository, ) + self.check_config_during_discover = self._requires_config_for_discover() + def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: """ Preprocesses the provided manifest dictionary by resolving any manifest references. @@ -978,3 +980,35 @@ def _select_streams( as_airbyte_message(configured_stream.stream, AirbyteStreamStatus.INCOMPLETE) ) return abstract_streams + + def _requires_config_for_discover(self) -> bool: + """Return whether `discover` needs a config to build this source's catalog. + + A config is required when the set of streams depends on the config (`ConditionalStreams`, + or `dynamic_streams` resolved from the API or the config rather than static parameters), or + when any stream schema is fetched from the API (`DynamicSchemaLoader`). + """ + for dynamic_definition in self._source_config.get("dynamic_streams", []): + resolver_type = dynamic_definition.get("components_resolver", {}).get("type") + if resolver_type != "ParametrizedComponentsResolver": + return True + + stream_definitions = [ + self._source_config.get("streams", []), + self._source_config.get("dynamic_streams", []), + ] + return any( + _contains_component_type(stream_definitions, component_type) + for component_type in ("ConditionalStreams", "DynamicSchemaLoader") + ) + + +def _contains_component_type(node: Any, component_type: str) -> bool: + """Return whether any mapping nested under `node` has the given component `type`.""" + if isinstance(node, Mapping): + return node.get("type") == component_type or any( + _contains_component_type(value, component_type) for value in node.values() + ) + if isinstance(node, list): + return any(_contains_component_type(item, component_type) for item in node) + return False diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source_config_free_discover.py b/unit_tests/sources/declarative/test_concurrent_declarative_source_config_free_discover.py new file mode 100644 index 0000000000..f08027619d --- /dev/null +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source_config_free_discover.py @@ -0,0 +1,184 @@ +# +# Copyright (c) 2025 Airbyte, Inc., all rights reserved. +# + +from typing import Any, Mapping + +import pytest + +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + ConcurrentDeclarativeSource, +) + +_RETRIEVER = { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url": "https://example.com/items", + "http_method": "GET", + "authenticator": {"type": "BearerAuthenticator", "api_token": "{{ config['api_key'] }}"}, + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, +} +_INLINE_SCHEMA_LOADER = { + "type": "InlineSchemaLoader", + "schema": {"type": "object", "properties": {"id": {"type": "string"}}}, +} +_DYNAMIC_SCHEMA_LOADER = { + "type": "DynamicSchemaLoader", + "retriever": _RETRIEVER, + "schema_type_identifier": {"type": "SchemaTypeIdentifier", "key_pointer": ["name"]}, +} + + +def _stream(name: str, schema_loader: Any) -> Mapping[str, Any]: + return { + "type": "DeclarativeStream", + "name": name, + "primary_key": [], + "schema_loader": schema_loader, + "retriever": _RETRIEVER, + } + + +def _dynamic_streams(resolver: Mapping[str, Any], schema_loader: Any) -> list[Mapping[str, Any]]: + return [ + { + "type": "DynamicDeclarativeStream", + "stream_template": _stream("template", schema_loader), + "components_resolver": resolver, + } + ] + + +_PARAMETRIZED_RESOLVER = { + "type": "ParametrizedComponentsResolver", + "stream_parameters": { + "type": "StreamParametersDefinition", + "list_of_parameters_for_stream": [{"name": "dynamic_items"}], + }, + "components_mapping": [ + { + "type": "ComponentMappingDefinition", + "field_path": ["name"], + "value": "{{ components_values['name'] }}", + } + ], +} +_HTTP_RESOLVER = { + "type": "HttpComponentsResolver", + "retriever": _RETRIEVER, + "components_mapping": _PARAMETRIZED_RESOLVER["components_mapping"], +} + + +def _manifest( + streams: list[Mapping[str, Any]], dynamic_streams: list[Mapping[str, Any]] | None = None +) -> Mapping[str, Any]: + manifest: dict[str, Any] = { + "version": "6.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["items"]}, + "streams": streams, + "spec": { + "type": "Spec", + "connection_specification": { + "type": "object", + "required": ["api_key"], + "properties": {"api_key": {"type": "string", "airbyte_secret": True}}, + }, + }, + } + if dynamic_streams: + manifest["dynamic_streams"] = dynamic_streams + return manifest + + +@pytest.mark.parametrize( + "manifest, expected_check_config_during_discover", + [ + pytest.param(_manifest([_stream("items", _INLINE_SCHEMA_LOADER)]), False, id="inline"), + pytest.param( + _manifest([_stream("items", _DYNAMIC_SCHEMA_LOADER)]), True, id="dynamic_schema_loader" + ), + pytest.param( + _manifest([_stream("items", [_INLINE_SCHEMA_LOADER, _DYNAMIC_SCHEMA_LOADER])]), + True, + id="dynamic_schema_loader_in_list", + ), + pytest.param( + _manifest( + [ + { + "type": "ConditionalStreams", + "condition": "{{ config.get('include_dynamic') }}", + "streams": [_stream("items", _DYNAMIC_SCHEMA_LOADER)], + } + ] + ), + True, + id="dynamic_schema_loader_in_inactive_conditional_streams", + ), + pytest.param( + _manifest( + [ + { + "type": "ConditionalStreams", + "condition": "{{ config.get('is_sandbox') }}", + "streams": [_stream("sandbox_items", _INLINE_SCHEMA_LOADER)], + } + ] + ), + True, + id="conditional_streams", + ), + pytest.param( + _manifest( + [_stream("items", _INLINE_SCHEMA_LOADER)], + _dynamic_streams(_PARAMETRIZED_RESOLVER, _INLINE_SCHEMA_LOADER), + ), + False, + id="parametrized_dynamic_streams", + ), + pytest.param( + _manifest( + [_stream("items", _INLINE_SCHEMA_LOADER)], + _dynamic_streams(_PARAMETRIZED_RESOLVER, _DYNAMIC_SCHEMA_LOADER), + ), + True, + id="parametrized_dynamic_streams_with_dynamic_schema_loader", + ), + pytest.param( + _manifest( + [_stream("items", _INLINE_SCHEMA_LOADER)], + _dynamic_streams(_HTTP_RESOLVER, _INLINE_SCHEMA_LOADER), + ), + True, + id="http_dynamic_streams", + ), + ], +) +def test_check_config_during_discover( + manifest: Mapping[str, Any], expected_check_config_during_discover: bool +) -> None: + source = ConcurrentDeclarativeSource(source_config=manifest, config={}) + + assert source.check_config_during_discover is expected_check_config_during_discover + assert source.check_config_against_spec is True + + +def test_discover_static_schema_without_config() -> None: + source = ConcurrentDeclarativeSource( + source_config=_manifest( + [_stream("items", _INLINE_SCHEMA_LOADER)], + _dynamic_streams(_PARAMETRIZED_RESOLVER, _INLINE_SCHEMA_LOADER), + ), + config={}, + ) + + catalog = source.discover(source.logger, {}) + + assert sorted(stream.name for stream in catalog.streams) == ["dynamic_items", "items"] diff --git a/unit_tests/test_entrypoint.py b/unit_tests/test_entrypoint.py index 58b94a85a6..1ff3a76308 100644 --- a/unit_tests/test_entrypoint.py +++ b/unit_tests/test_entrypoint.py @@ -238,14 +238,13 @@ def test_parse_new_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] @@ -1003,3 +1002,116 @@ def test_check_masks_secrets_in_a_config_error_raised_by_the_connector( assert "secretpw" not in "".join(messages) assert orjson.loads(messages[-1])["connectionStatus"]["message"] == "Login failed for ****" + + +def _mock_message_repository(mocker): + message_repository = MagicMock() + message_repository.consume_queue.return_value = [] + mocker.patch.object( + MockSource, + "message_repository", + new_callable=mocker.PropertyMock, + return_value=message_repository, + ) + + +def test_run_discover_without_config_when_supported(mocker): + _mock_message_repository(mocker) + source = MockSource() + source.check_config_during_discover = False + expected_catalog = AirbyteCatalog( + streams=[ + AirbyteStream( + name="test_stream", + json_schema={"type": "object"}, + supported_sync_modes=[SyncMode.full_refresh], + ) + ] + ) + mocker.patch.object( + MockSource, "spec", return_value=ConnectorSpecification(connectionSpecification={}) + ) + mocker.patch.object(MockSource, "discover", return_value=expected_catalog) + + messages = list(AirbyteEntrypoint(source).run(Namespace(command="discover", config=None))) + + assert messages == [_wrap_message(expected_catalog)] + assert MockSource.discover.call_args[0][1] == {} + + +def test_run_discover_without_config_when_not_supported(mocker): + _mock_message_repository(mocker) + source = MockSource() + mocker.patch.object( + MockSource, "spec", return_value=ConnectorSpecification(connectionSpecification={}) + ) + discover = mocker.patch.object(MockSource, "discover") + + with pytest.raises(AirbyteTracedException) as exc_info: + list(AirbyteEntrypoint(source).run(Namespace(command="discover", config=None))) + + assert exc_info.value.failure_type == FailureType.config_error + assert "--config" in exc_info.value.message + discover.assert_not_called() + + +def test_run_discover_without_config_wraps_source_errors(mocker): + _mock_message_repository(mocker) + source = MockSource() + source.check_config_during_discover = False + mocker.patch.object( + MockSource, "spec", return_value=ConnectorSpecification(connectionSpecification={}) + ) + mocker.patch.object(MockSource, "discover", side_effect=KeyError("start_date")) + + with pytest.raises(AirbyteTracedException) as exc_info: + list(AirbyteEntrypoint(source).run(Namespace(command="discover", config=None))) + + assert exc_info.value.failure_type == FailureType.config_error + assert "start_date" in exc_info.value.internal_message + + +def test_run_discover_with_config_skips_validation_when_check_config_against_spec_is_false( + mocker, +): + _mock_message_repository(mocker) + source = MockSource() + source.check_config_against_spec = False + mocker.patch.object( + MockSource, + "spec", + return_value=ConnectorSpecification( + connectionSpecification={"type": "object", "required": ["api_key"]} + ), + ) + mocker.patch.object(MockSource, "read_config", return_value={}) + mocker.patch.object(MockSource, "configure", return_value={}) + catalog = AirbyteCatalog(streams=[]) + mocker.patch.object(MockSource, "discover", return_value=catalog) + + messages = list( + AirbyteEntrypoint(source).run(Namespace(command="discover", config="config_path")) + ) + + assert messages == [_wrap_message(catalog)] + + +def test_run_discover_with_config_validates_config_when_config_free_discover_is_supported(mocker): + _mock_message_repository(mocker) + source = MockSource() + source.check_config_during_discover = False + mocker.patch.object( + MockSource, + "spec", + return_value=ConnectorSpecification( + connectionSpecification={"type": "object", "required": ["api_key"]} + ), + ) + mocker.patch.object(MockSource, "read_config", return_value={}) + mocker.patch.object(MockSource, "configure", return_value={}) + discover = mocker.patch.object(MockSource, "discover") + + with pytest.raises(AirbyteTracedException): + list(AirbyteEntrypoint(source).run(Namespace(command="discover", config="config_path"))) + + discover.assert_not_called()