diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index e227eca34..4b8f52f23 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -4,6 +4,7 @@ from __future__ import annotations import hashlib +import json import warnings from pathlib import Path from typing import IO, TYPE_CHECKING, Any, cast @@ -38,6 +39,27 @@ def _suppress_cdk_pydantic_deprecation_warnings() -> None: ) +def _get_config_from_args(args: list[str]) -> dict[str, Any]: + config_path: str | None = None + try: + config_path = args[args.index("--config") + 1] + except (IndexError, ValueError): + for arg in args: + if arg.startswith("--config="): + config_path = arg.partition("=")[2] + break + + if not config_path: + return {} + + try: + config = json.loads(Path(config_path).read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return {} + + return config if isinstance(config, dict) else {} + + class DeclarativeExecutor(Executor): """An executor for declarative sources.""" @@ -45,6 +67,8 @@ def __init__( self, name: str, manifest: dict | Path, + *, + config: dict[str, Any] | None = None, components_py: str | Path | None = None, components_py_checksum: str | None = None, ) -> None: @@ -53,6 +77,7 @@ def __init__( - If `manifest` is a path, it will be read as a json file. - If `manifest` is a string, it will be parsed as an HTTP path. - If `manifest` is a dict, it will be used as is. + - If `config` is provided, it will be used to resolve manifest interpolations. - If `components_py` is provided, components will be injected into the source. - If `components_py_checksum` is not provided, it will be calculated automatically. """ @@ -66,7 +91,7 @@ def __init__( elif isinstance(manifest, dict): self._manifest_dict = manifest - config_dict: dict[str, Any] = {} + config_dict: dict[str, Any] = dict(config or {}) if components_py: if isinstance(components_py, Path): components_py = components_py.read_text() @@ -82,6 +107,25 @@ def __init__( self.reported_version: str | None = self._manifest_dict.get("version", None) self._config_dict = config_dict + def _create_declarative_source( + self, + config: dict[str, Any], + ) -> ConcurrentDeclarativeSource: + return ConcurrentDeclarativeSource( + config=config, + source_config=self._manifest_dict, + ) + + def _get_effective_config(self, args_config: dict[str, Any]) -> dict[str, Any]: + config = {**self._config_dict, **args_config} + for key in ( + "__injected_components_py", + "__injected_components_py_checksums", + ): + if key in self._config_dict: + config[key] = self._config_dict[key] + return config + @property def declarative_source(self) -> ConcurrentDeclarativeSource: """Get the declarative source object. @@ -93,10 +137,7 @@ def declarative_source(self) -> ConcurrentDeclarativeSource: 3. Rather than cache the source object, we recreate it each time we need it, to avoid any issues with re-using the same object. """ - return ConcurrentDeclarativeSource( - config=self._config_dict, - source_config=self._manifest_dict, - ) + return self._create_declarative_source(self._config_dict) def get_installed_version( self, @@ -122,9 +163,11 @@ def execute( ) -> Iterator[str]: """Execute the declarative source.""" _ = stdin, suppress_stderr # Not used - source_entrypoint = AirbyteEntrypoint(self.declarative_source) - mapped_args: list[str] = self.map_cli_args(args) + args_config = _get_config_from_args(mapped_args) + source_entrypoint = AirbyteEntrypoint( + self._create_declarative_source(self._get_effective_config(args_config)) + ) parsed_args: Namespace = source_entrypoint.parse_args(mapped_args) yield from source_entrypoint.run(parsed_args) diff --git a/airbyte/_executors/util.py b/airbyte/_executors/util.py index 7a30a1bf1..dc808948d 100644 --- a/airbyte/_executors/util.py +++ b/airbyte/_executors/util.py @@ -7,7 +7,7 @@ import tempfile import zipfile from pathlib import Path -from typing import TYPE_CHECKING, Literal, cast +from typing import TYPE_CHECKING, Any, Literal, cast import requests import yaml @@ -184,10 +184,12 @@ def get_connector_executor( # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 # install_root: Path | None = None, use_python: bool | Path | str | None = None, no_executor: bool = False, + config: dict[str, Any] | None = None, ) -> Executor: """This factory function creates an executor for a connector. - For documentation of each arg, see the function `airbyte.sources.util.get_source()`. + For documentation of each arg, see the function `airbyte.sources.util.get_source()`. The + `config` argument is also used to resolve declarative manifest interpolations. """ install_method_count = sum( [ @@ -349,6 +351,7 @@ def get_connector_executor( # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 # return DeclarativeExecutor( name=name, manifest=source_manifest, + config=config, components_py=components_py_path, ) @@ -364,6 +367,7 @@ def get_connector_executor( # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 # return DeclarativeExecutor( name=name, manifest=manifest_dict, + config=config, components_py=components_py, components_py_checksum=components_py_checksum, ) diff --git a/airbyte/sources/util.py b/airbyte/sources/util.py index 42372ed03..96f2cbd11 100644 --- a/airbyte/sources/util.py +++ b/airbyte/sources/util.py @@ -128,6 +128,7 @@ def get_source( # noqa: PLR0913 # Too many arguments install_if_missing=install_if_missing, install_root=install_root, no_executor=no_executor, + config=config, ) return Source( diff --git a/tests/unit_tests/test_declarative_executor.py b/tests/unit_tests/test_declarative_executor.py new file mode 100644 index 000000000..96a4ae929 --- /dev/null +++ b/tests/unit_tests/test_declarative_executor.py @@ -0,0 +1,127 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +"""Unit tests for declarative executor configuration resolution.""" + +from __future__ import annotations + +from collections.abc import Iterator +import json +from pathlib import Path +from typing import Any + +import pytest + +from airbyte._executors import declarative +from airbyte._executors.declarative import DeclarativeExecutor +from airbyte.sources import util as sources_util + + +def test_get_source_passes_config_to_declarative_executor( + monkeypatch: pytest.MonkeyPatch, +) -> None: + manifest = {"version": "1.0.0"} + config = {"api_key": "configured"} + + def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: + return DeclarativeExecutor( + name=kwargs["name"], + manifest=kwargs["source_manifest"], + config=kwargs["config"], + components_py="class Component:\n pass\n", + ) + + monkeypatch.setattr( + sources_util, + "get_connector_executor", + fake_get_connector_executor, + ) + monkeypatch.setattr( + declarative, + "ConcurrentDeclarativeSource", + lambda **kwargs: kwargs, + ) + + source = sources_util.get_source( + name="source-test", + config=config, + source_manifest=manifest, + ) + + declarative_config = source.executor.declarative_source["config"] + assert declarative_config["api_key"] == config["api_key"] + assert declarative_config["__injected_components_py"] + assert config == {"api_key": "configured"} + + +def test_execute_uses_config_set_after_get_source_and_preserves_injected_components( + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, +) -> None: + captured: dict[str, Any] = {} + manifest = {"version": "1.0.0"} + late_config = {"api_key": "configured-later"} + + def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: + return DeclarativeExecutor( + name=kwargs["name"], + manifest=kwargs["source_manifest"], + config=kwargs["config"], + components_py="class Component:\n pass\n", + ) + + class FakeEntrypoint: + def __init__(self, source: Any) -> None: + captured["source"] = source + + def parse_args(self, args: list[str]) -> list[str]: + return args + + def run(self, args: list[str]) -> Iterator[str]: + yield from args + + monkeypatch.setattr( + sources_util, + "get_connector_executor", + fake_get_connector_executor, + ) + monkeypatch.setattr( + declarative, + "ConcurrentDeclarativeSource", + lambda **kwargs: kwargs, + ) + monkeypatch.setattr(declarative, "AirbyteEntrypoint", FakeEntrypoint) + + source = sources_util.get_source( + name="source-test", + source_manifest=manifest, + ) + source.set_config(late_config, validate=False) + config_path = tmp_path / "config.json" + config_file_config = { + **source._hydrated_config, + "__injected_components_py": "bogus", + "__injected_components_py_checksums": {"md5": "bogus"}, + } + config_path.write_text( + json.dumps(config_file_config), + encoding="utf-8", + ) + + list(source.executor.execute(["read", "--config", str(config_path)])) + + config = captured["source"]["config"] + assert config["api_key"] == late_config["api_key"] + assert config["__injected_components_py"] == "class Component:\n pass\n" + assert config["__injected_components_py_checksums"]["md5"] + + +def test_declarative_executor_copies_config_before_component_injection() -> None: + config = {"api_key": "configured"} + + DeclarativeExecutor( + name="source-test", + manifest={"version": "1.0.0"}, + config=config, + components_py="class Component:\n pass\n", + ) + + assert config == {"api_key": "configured"}