From 5f43def937e88a1a8c25beb9310b17bc67f5451e Mon Sep 17 00:00:00 2001 From: Adileo Barone Date: Sun, 16 Nov 2025 18:22:31 +0100 Subject: [PATCH 01/13] Use provided config in declarative executor --- airbyte/_executors/declarative.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index e227eca34..e71b85ace 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -45,6 +45,7 @@ def __init__( self, name: str, manifest: dict | Path, + config: dict[str, Any] = {}, components_py: str | Path | None = None, components_py_checksum: str | None = None, ) -> None: @@ -66,7 +67,7 @@ def __init__( elif isinstance(manifest, dict): self._manifest_dict = manifest - config_dict: dict[str, Any] = {} + config_dict: dict[str, Any] = config if components_py: if isinstance(components_py, Path): components_py = components_py.read_text() From fcf319783da1041f4dcc76a6b40cdcc761fd512b Mon Sep 17 00:00:00 2001 From: Adileo Barone Date: Sun, 16 Nov 2025 18:26:50 +0100 Subject: [PATCH 02/13] Add config parameter to executor factory function --- airbyte/_executors/util.py | 3 +++ 1 file changed, 3 insertions(+) diff --git a/airbyte/_executors/util.py b/airbyte/_executors/util.py index 7a30a1bf1..150272caa 100644 --- a/airbyte/_executors/util.py +++ b/airbyte/_executors/util.py @@ -184,6 +184,7 @@ 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] = {} ) -> Executor: """This factory function creates an executor for a connector. @@ -349,6 +350,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 +366,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, ) From 2a118e9ad2f3392feb6a9ca8a755ba3738d850ca Mon Sep 17 00:00:00 2001 From: Adileo Barone Date: Sun, 16 Nov 2025 18:27:52 +0100 Subject: [PATCH 03/13] Add config parameter to Source initialization --- airbyte/sources/util.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airbyte/sources/util.py b/airbyte/sources/util.py index 42372ed03..82f2c80a2 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( From 33256c08a66404e87abb5aece1677bdff3088009 Mon Sep 17 00:00:00 2001 From: "Aaron (\"AJ\") Steers" Date: Wed, 26 Nov 2025 19:22:16 -0800 Subject: [PATCH 04/13] Apply suggestion from @coderabbitai[bot] Co-authored-by: coderabbitai[bot] <136622811+coderabbitai[bot]@users.noreply.github.com> --- airbyte/sources/util.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/airbyte/sources/util.py b/airbyte/sources/util.py index 82f2c80a2..96f2cbd11 100644 --- a/airbyte/sources/util.py +++ b/airbyte/sources/util.py @@ -128,7 +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 + config=config, ) return Source( From d734058ee502aee4874366b33992db1ab3cd1743 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:25:29 +0000 Subject: [PATCH 05/13] fix: avoid mutable declarative config defaults Co-Authored-By: AJ Steers --- airbyte/_executors/declarative.py | 5 +++-- airbyte/_executors/util.py | 5 ++++- 2 files changed, 7 insertions(+), 3 deletions(-) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index e71b85ace..11f6b7aa6 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -45,7 +45,7 @@ def __init__( self, name: str, manifest: dict | Path, - config: dict[str, Any] = {}, + config: dict[str, Any] | None = None, components_py: str | Path | None = None, components_py_checksum: str | None = None, ) -> None: @@ -54,6 +54,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. """ @@ -67,7 +68,7 @@ def __init__( elif isinstance(manifest, dict): self._manifest_dict = manifest - config_dict: dict[str, Any] = config + config_dict: dict[str, Any] = dict(config or {}) if components_py: if isinstance(components_py, Path): components_py = components_py.read_text() diff --git a/airbyte/_executors/util.py b/airbyte/_executors/util.py index 150272caa..29819705e 100644 --- a/airbyte/_executors/util.py +++ b/airbyte/_executors/util.py @@ -184,11 +184,14 @@ 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] = {} + 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()`. + + Args: + config: Connector config used to resolve declarative manifest interpolations. """ install_method_count = sum( [ From d7da9c8c14dce2896e3f3ecdd3a4f6b87f6ccad6 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:25:40 +0000 Subject: [PATCH 06/13] fix: resolve declarative config from execution args Co-Authored-By: AJ Steers --- airbyte/_executors/declarative.py | 31 +++++++++++++++++++++++++++++-- 1 file changed, 29 insertions(+), 2 deletions(-) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index 11f6b7aa6..d86ac1f0e 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()) + except (OSError, UnicodeError, json.JSONDecodeError): + return {} + + return config if isinstance(config, dict) else {} + + class DeclarativeExecutor(Executor): """An executor for declarative sources.""" @@ -124,9 +146,14 @@ 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( + ConcurrentDeclarativeSource( + config={**self._config_dict, **args_config}, + source_config=self._manifest_dict, + ) + ) parsed_args: Namespace = source_entrypoint.parse_args(mapped_args) yield from source_entrypoint.run(parsed_args) From d42c363299e337be932f0616968265938a3587cc Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:26:08 +0000 Subject: [PATCH 07/13] test: cover declarative config resolution paths Co-Authored-By: AJ Steers --- tests/unit_tests/test_declarative_executor.py | 109 ++++++++++++++++++ 1 file changed, 109 insertions(+) create mode 100644 tests/unit_tests/test_declarative_executor.py diff --git a/tests/unit_tests/test_declarative_executor.py b/tests/unit_tests/test_declarative_executor.py new file mode 100644 index 000000000..bc4015207 --- /dev/null +++ b/tests/unit_tests/test_declarative_executor.py @@ -0,0 +1,109 @@ +from __future__ import annotations + +import json +from typing import Any + +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) -> None: + captured: dict[str, Any] = {} + manifest = {"version": "1.0.0"} + config = {"api_key": "configured"} + + def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: + captured.update(kwargs) + return DeclarativeExecutor( + name=kwargs["name"], + manifest=kwargs["source_manifest"], + config=kwargs["config"], + ) + + 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, + ) + + assert source.executor.declarative_source["config"] == config + assert captured["config"] == config + + +def test_execute_uses_config_set_after_get_source_and_preserves_injected_components( + monkeypatch, + tmp_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]): + 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_path.write_text(json.dumps(source._hydrated_config)) + + 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"} From 3fb2f86c0194f3236627bc16cee0d0a82dbd65ff Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:26:46 +0000 Subject: [PATCH 08/13] test: cover component injection during source creation Co-Authored-By: AJ Steers --- tests/unit_tests/test_declarative_executor.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/unit_tests/test_declarative_executor.py b/tests/unit_tests/test_declarative_executor.py index bc4015207..858aeb4bb 100644 --- a/tests/unit_tests/test_declarative_executor.py +++ b/tests/unit_tests/test_declarative_executor.py @@ -19,6 +19,7 @@ def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: name=kwargs["name"], manifest=kwargs["source_manifest"], config=kwargs["config"], + components_py="class Component:\n pass\n", ) monkeypatch.setattr( @@ -40,6 +41,7 @@ def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: assert source.executor.declarative_source["config"] == config assert captured["config"] == config + assert config == {"api_key": "configured"} def test_execute_uses_config_set_after_get_source_and_preserves_injected_components( From 27f89ddcf1f000d6aeb25c48babfcb969a145b22 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:27:10 +0000 Subject: [PATCH 09/13] fix: satisfy declarative config lint checks Co-Authored-By: AJ Steers --- airbyte/_executors/declarative.py | 2 +- airbyte/_executors/util.py | 8 +++----- tests/unit_tests/test_declarative_executor.py | 4 +++- 3 files changed, 7 insertions(+), 7 deletions(-) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index d86ac1f0e..a538d200e 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -53,7 +53,7 @@ def _get_config_from_args(args: list[str]) -> dict[str, Any]: return {} try: - config = json.loads(Path(config_path).read_text()) + config = json.loads(Path(config_path).read_text(encoding="utf-8")) except (OSError, UnicodeError, json.JSONDecodeError): return {} diff --git a/airbyte/_executors/util.py b/airbyte/_executors/util.py index 29819705e..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 @@ -188,10 +188,8 @@ def get_connector_executor( # noqa: PLR0912, PLR0913, PLR0914, PLR0915, C901 # ) -> Executor: """This factory function creates an executor for a connector. - For documentation of each arg, see the function `airbyte.sources.util.get_source()`. - - Args: - config: Connector config used to resolve declarative manifest interpolations. + 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( [ diff --git a/tests/unit_tests/test_declarative_executor.py b/tests/unit_tests/test_declarative_executor.py index 858aeb4bb..3b3b37c48 100644 --- a/tests/unit_tests/test_declarative_executor.py +++ b/tests/unit_tests/test_declarative_executor.py @@ -39,7 +39,9 @@ def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: source_manifest=manifest, ) - assert source.executor.declarative_source["config"] == config + declarative_config = source.executor.declarative_source["config"] + assert declarative_config["api_key"] == config["api_key"] + assert declarative_config["__injected_components_py"] assert captured["config"] == config assert config == {"api_key": "configured"} From 3dd0ee1b46dff7b4f767e079591c9f8f7d996c49 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:30:20 +0000 Subject: [PATCH 10/13] fix: preserve declarative executor call compatibility Co-Authored-By: AJ Steers --- airbyte/_executors/declarative.py | 1 + 1 file changed, 1 insertion(+) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index a538d200e..6dcbf0479 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -67,6 +67,7 @@ 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, From 097973561a26f729aaa87a319f4d5474da971836 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:30:34 +0000 Subject: [PATCH 11/13] refactor: centralize declarative source creation Co-Authored-By: AJ Steers --- airbyte/_executors/declarative.py | 19 ++++++++++++------- 1 file changed, 12 insertions(+), 7 deletions(-) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index 6dcbf0479..d762ee88a 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -107,6 +107,15 @@ 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, + ) + @property def declarative_source(self) -> ConcurrentDeclarativeSource: """Get the declarative source object. @@ -118,10 +127,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, @@ -150,9 +156,8 @@ def execute( mapped_args: list[str] = self.map_cli_args(args) args_config = _get_config_from_args(mapped_args) source_entrypoint = AirbyteEntrypoint( - ConcurrentDeclarativeSource( - config={**self._config_dict, **args_config}, - source_config=self._manifest_dict, + self._create_declarative_source( + {**self._config_dict, **args_config}, ) ) parsed_args: Namespace = source_entrypoint.parse_args(mapped_args) From b3dd8125ee56042bcfff5e2ae80b7366875d8208 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:30:53 +0000 Subject: [PATCH 12/13] test: align declarative executor test conventions Co-Authored-By: AJ Steers --- tests/unit_tests/test_declarative_executor.py | 25 +++++++++++++------ 1 file changed, 17 insertions(+), 8 deletions(-) diff --git a/tests/unit_tests/test_declarative_executor.py b/tests/unit_tests/test_declarative_executor.py index 3b3b37c48..b6272baf6 100644 --- a/tests/unit_tests/test_declarative_executor.py +++ b/tests/unit_tests/test_declarative_executor.py @@ -1,20 +1,27 @@ +# 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) -> None: - captured: dict[str, Any] = {} +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: - captured.update(kwargs) return DeclarativeExecutor( name=kwargs["name"], manifest=kwargs["source_manifest"], @@ -42,13 +49,12 @@ def fake_get_connector_executor(**kwargs: Any) -> DeclarativeExecutor: declarative_config = source.executor.declarative_source["config"] assert declarative_config["api_key"] == config["api_key"] assert declarative_config["__injected_components_py"] - assert captured["config"] == config assert config == {"api_key": "configured"} def test_execute_uses_config_set_after_get_source_and_preserves_injected_components( - monkeypatch, - tmp_path, + monkeypatch: pytest.MonkeyPatch, + tmp_path: Path, ) -> None: captured: dict[str, Any] = {} manifest = {"version": "1.0.0"} @@ -69,7 +75,7 @@ def __init__(self, source: Any) -> None: def parse_args(self, args: list[str]) -> list[str]: return args - def run(self, args: list[str]): + def run(self, args: list[str]) -> Iterator[str]: yield from args monkeypatch.setattr( @@ -90,7 +96,10 @@ def run(self, args: list[str]): ) source.set_config(late_config, validate=False) config_path = tmp_path / "config.json" - config_path.write_text(json.dumps(source._hydrated_config)) + config_path.write_text( + json.dumps(source._hydrated_config), + encoding="utf-8", + ) list(source.executor.execute(["read", "--config", str(config_path)])) From 95d7aaec4a55ca046f6b3d64c70bbdb153bf2dc6 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Tue, 18 Aug 2026 02:38:55 +0000 Subject: [PATCH 13/13] fix: preserve declarative component injection Co-Authored-By: AJ Steers --- airbyte/_executors/declarative.py | 14 +++++++++++--- tests/unit_tests/test_declarative_executor.py | 7 ++++++- 2 files changed, 17 insertions(+), 4 deletions(-) diff --git a/airbyte/_executors/declarative.py b/airbyte/_executors/declarative.py index d762ee88a..4b8f52f23 100644 --- a/airbyte/_executors/declarative.py +++ b/airbyte/_executors/declarative.py @@ -116,6 +116,16 @@ def _create_declarative_source( 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. @@ -156,9 +166,7 @@ def execute( 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._config_dict, **args_config}, - ) + 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/tests/unit_tests/test_declarative_executor.py b/tests/unit_tests/test_declarative_executor.py index b6272baf6..96a4ae929 100644 --- a/tests/unit_tests/test_declarative_executor.py +++ b/tests/unit_tests/test_declarative_executor.py @@ -96,8 +96,13 @@ def run(self, args: list[str]) -> Iterator[str]: ) 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(source._hydrated_config), + json.dumps(config_file_config), encoding="utf-8", )