fix(declarative): [DO-NOT-MERGE] pass config into declarative source

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,13 +39,36 @@ 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."""
 
     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)
 

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,
             )

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(

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"}