From 29fb89a1362f752d87957634775967637b648015 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 19 Aug 2026 13:29:52 +0300 Subject: [PATCH 1/6] feat(low-code): let the check component override config values A check and a sync legitimately want different behaviour from the same manifest. A check is interactive and should fail fast with an actionable message; a sync can afford to wait out a rate limit window. Today the only way to express that difference is a Python `check_connection` override that builds a second component tree from a modified config, which a manifest-only connector cannot do. `CheckStream` and `CheckDynamicStream` gain an optional `config_overrides` mapping. `check()` overlays it onto the config for the duration of the check, which is enough to reach every component the checker builds because `streams()` interpolates from `self._config` and ignores its own `config` argument. Values are applied verbatim and are not interpolated. `config_validations` continue to run against the config the user supplied, so an override cannot fail a validation the user has no way to satisfy. Inert by default: with no `config_overrides` key, `check()` behaves exactly as before. Co-Authored-By: Claude Opus 5 (1M context) --- .../concurrent_declarative_source.py | 38 +++- .../declarative_component_schema.yaml | 16 ++ .../models/declarative_component_schema.py | 12 ++ .../declarative/checks/test_check_stream.py | 183 ++++++++++++++++++ 4 files changed, 247 insertions(+), 2 deletions(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index d7e5e4260..e6faa929b 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -3,6 +3,7 @@ import json import logging import pkgutil +from contextlib import contextmanager from copy import deepcopy from dataclasses import dataclass, field from queue import Queue @@ -228,6 +229,10 @@ def __init__( self._constructor.create_component(SpecModel, spec, dict()) if spec else None ) self._config = self._migrate_and_transform_config(config_path, config) or {} + # `check` may temporarily overlay values onto `self._config` (see + # `_config_overridden_for_check`). The manifest's `config_validations` express intent about what + # the *user* supplied, so they must always run against the unmodified config. + self._user_provided_config = self._config concurrency_level_from_manifest = self._source_config.get("concurrency_level") if concurrency_level_from_manifest: @@ -413,7 +418,7 @@ def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]: # type: i """ if self._spec_component: - self._spec_component.validate_config(self._config) + self._spec_component.validate_config(self._user_provided_config) api_budget_model = self._source_config.get("api_budget") if api_budget_model: @@ -607,11 +612,40 @@ def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCon f"Expected to generate a ConnectionChecker component, but received {connection_checker.__class__}" ) - check_succeeded, error = connection_checker.check_connection(self, logger, self._config) + with self._config_overridden_for_check(check.get("config_overrides")): + check_succeeded, error = connection_checker.check_connection(self, logger, self._config) if not check_succeeded: return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error)) return AirbyteConnectionStatus(status=Status.SUCCEEDED) + @contextmanager + def _config_overridden_for_check( + self, config_overrides: Optional[Mapping[str, Any]] + ) -> Iterator[None]: + """Overlay the check component's `config_overrides` onto the config for the duration of a check. + + A check and a sync legitimately want different behaviour from the same manifest: a check is + interactive and should fail fast with an actionable message, while a sync can afford to wait out + a rate limit window. Before this existed, expressing that difference required a Python + `check_connection` override that built a second component tree from a modified config, which is + not available to a manifest-only connector. + + Overlaying here is enough to reach every component the checker builds, because `streams()` + interpolates from `self._config` and ignores its own `config` argument. Values are applied + verbatim - they are not interpolated - and `config_validations` still run against the config the + user supplied, so an override cannot fail a validation the user has no way to satisfy. + """ + if not config_overrides: + yield + return + + unmodified_config = self._config + self._config = {**self._config, **config_overrides} + try: + yield + finally: + self._config = unmodified_config + @property def dynamic_streams(self) -> List[Dict[str, Any]]: return self._dynamic_stream_configs( diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 091c694d2..ebedd71b7 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -549,6 +549,14 @@ definitions: type: array items: "$ref": "#/definitions/DynamicStreamCheckConfig" + config_overrides: + title: Config Overrides + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated. + type: object + additionalProperties: true + examples: + - max_waiting_time: 0 + - page_size: 1 DynamicStreamCheckConfig: type: object required: @@ -587,6 +595,14 @@ definitions: description: Enables stream check availability. This field is automatically set by the CDK. type: boolean default: true + config_overrides: + title: Config Overrides + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated. + type: object + additionalProperties: true + examples: + - max_waiting_time: 0 + - page_size: 1 CompositeErrorHandler: title: Composite Error Handler description: Error handler that sequentially iterates over a list of error handlers. diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 37eb50b7b..b20cdb4e4 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -77,6 +77,12 @@ class CheckDynamicStream(BaseModel): description="Enables stream check availability. This field is automatically set by the CDK.", title="Use Check Availability", ) + config_overrides: Optional[Dict[str, Any]] = Field( + None, + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated.", + examples=[{"max_waiting_time": 0}, {"page_size": 1}], + title="Config Overrides", + ) class ConcurrencyLevel(BaseModel): @@ -1781,6 +1787,12 @@ class CheckStream(BaseModel): title="Stream Names", ) dynamic_streams_check_configs: Optional[List[DynamicStreamCheckConfig]] = None + config_overrides: Optional[Dict[str, Any]] = Field( + None, + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated.", + examples=[{"max_waiting_time": 0}, {"page_size": 1}], + title="Config Overrides", + ) class IncrementingCountCursor(BaseModel): diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 8bc9571d1..9a488de5a 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -907,3 +907,186 @@ def test_check_stream_only_type_provided(): ) with pytest.raises(ValueError): source.check(logger, _CONFIG) + + +_CONFIG_DRIVEN_PATH_CONFIG = {"resource": "sync"} + +_MANIFEST_WITH_CONFIG_DRIVEN_PATH = { + "version": "6.7.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": ["items"]}, + "streams": [ + { + "type": "DeclarativeStream", + "name": "items", + "primary_key": "id", + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "$schema": "http://json-schema.org/schema#", + "type": "object", + "properties": {"id": {"type": "integer"}}, + }, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url": "https://api.test.com/{{ config['resource'] }}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "paginator": {"type": "NoPagination"}, + }, + } + ], +} + + +def _source_with_check_component(check_component): + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = check_component + return ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + +def test_given_no_config_overrides_when_check_then_components_use_the_user_config(): + source = _source_with_check_component({"type": "CheckStream", "stream_names": ["items"]}) + + with HttpMocker() as http_mocker: + # Only the user-configured path is mocked, so a request to any other path fails the test. + http_mocker.get( + HttpRequest(url="https://api.test.com/sync"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_config_overrides_when_check_then_components_built_during_check_see_them(): + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + + with HttpMocker() as http_mocker: + # Only the overridden path is mocked. Reaching this endpoint proves the overlay was applied to + # the stream the checker built, and not merely stored on the source. + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(overridden_request, 1) + + +def test_given_config_overrides_when_check_then_the_config_is_restored_afterwards(): + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + assert source._config == _CONFIG_DRIVEN_PATH_CONFIG + + # A sync that follows a check in the same process must go back to the user's value. + with HttpMocker() as http_mocker: + sync_request = HttpRequest(url="https://api.test.com/sync") + http_mocker.get(sync_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + stream = source.streams(_CONFIG_DRIVEN_PATH_CONFIG)[0] + assert stream.check_availability().is_available + + http_mocker.assert_number_of_calls(sync_request, 1) + + +def test_given_config_overrides_when_check_raises_then_the_config_is_restored(): + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["not_in_the_catalog"], + "config_overrides": {"resource": "check-only"}, + } + ) + + with pytest.raises(ValueError): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + assert source._config == _CONFIG_DRIVEN_PATH_CONFIG + + +def test_given_config_overrides_when_check_then_config_validations_run_against_the_user_config(): + """An override is authored in the manifest, so it must not be held to validations written for the + user's own input - the user has no way to satisfy them.""" + config = {"resource": "sync", "settings": {"mode": "sync"}} + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only", "settings": {"mode": "check-only"}}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "resource": {"type": "string"}, + "settings": {"type": "object"}, + }, + }, + "config_normalization_rules": { + "type": "ConfigNormalizationRules", + "validations": [ + { + "type": "DpathValidator", + "field_path": ["settings"], + "validation_strategy": { + "type": "ValidateAdheresToSchema", + "base_schema": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"mode": {"type": "string", "enum": ["sync"]}}, + "required": ["mode"], + "additionalProperties": False, + }, + }, + } + ], + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=config, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + # `settings.mode` is overridden to a value outside the validator's enum, so this would fail were + # the overlay validated instead of the config the user supplied. + assert source.check(logger, config).status == Status.SUCCEEDED From 0e23e0479eb83025fed3008d248e67f342379eee Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 19 Aug 2026 14:09:00 +0300 Subject: [PATCH 2/6] fix(low-code): document and pin config_overrides semantics Follow-up on the same branch, addressing a local review of the feature. Documents three semantics the field description and docstring left implicit: overrides are merged one level deep so a nested object is replaced rather than deep-merged; they are applied after config migrations and transformations, so an override is not normalised and derived fields are not recomputed; and the overlay mutates shared state, which is safe only because check is one command per process. Renames `_user_provided_config` to `_config_for_validation`. It holds the config after migrations and transformations have run, not what the user typed, and the old name plus its comment invited the wrong reading. Adds tests for two contracts that were stated but unpinned: `CheckDynamicStream` is covered by the overlay, and override values are verbatim rather than interpolated. Co-Authored-By: Claude Opus 5 (1M context) --- .../concurrent_declarative_source.py | 28 ++++-- .../declarative_component_schema.yaml | 4 +- .../models/declarative_component_schema.py | 4 +- .../declarative/checks/test_check_stream.py | 99 +++++++++++++++++++ 4 files changed, 124 insertions(+), 11 deletions(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index e6faa929b..5d67501e0 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -230,9 +230,9 @@ def __init__( ) self._config = self._migrate_and_transform_config(config_path, config) or {} # `check` may temporarily overlay values onto `self._config` (see - # `_config_overridden_for_check`). The manifest's `config_validations` express intent about what - # the *user* supplied, so they must always run against the unmodified config. - self._user_provided_config = self._config + # `_config_overridden_for_check`). The manifest's `config_validations` express intent about the + # config as supplied, so they must run against it rather than against a check-time overlay. + self._config_for_validation = self._config concurrency_level_from_manifest = self._source_config.get("concurrency_level") if concurrency_level_from_manifest: @@ -418,7 +418,7 @@ def streams(self, config: Mapping[str, Any]) -> List[AbstractStream]: # type: i """ if self._spec_component: - self._spec_component.validate_config(self._user_provided_config) + self._spec_component.validate_config(self._config_for_validation) api_budget_model = self._source_config.get("api_budget") if api_budget_model: @@ -631,9 +631,23 @@ def _config_overridden_for_check( not available to a manifest-only connector. Overlaying here is enough to reach every component the checker builds, because `streams()` - interpolates from `self._config` and ignores its own `config` argument. Values are applied - verbatim - they are not interpolated - and `config_validations` still run against the config the - user supplied, so an override cannot fail a validation the user has no way to satisfy. + interpolates from `self._config` and ignores its own `config` argument. + + Semantics, all deliberate: + - Values are applied verbatim. They are not interpolated, so a value containing `{{ }}` reaches + components as that literal string. + - The merge is one level deep. Overriding a key whose value is an object replaces that object + rather than merging into it, which keeps removing a nested key expressible. + - The overlay happens long after `_migrate_and_transform_config`, so an override is neither + normalised itself nor propagated to fields derived from it by a `ConfigTransformation`. + - `config_validations` run against `self._config_for_validation`, so an override cannot fail a + validation written for the config as supplied. + - `self._config` is shared state. Reassigning it is not thread safe, and a component that writes + back into the config during check - a `SingleUseRefreshTokenOauth2Authenticator` refreshing its + token, say - writes into the throwaway overlay, which the restore then discards. The control + message is still emitted so the platform persists the new token, but an in-process read + following a check would carry the stale one. Both are acceptable while this is scoped to + `check`, which is one command per process. """ if not config_overrides: yield diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index ebedd71b7..7506806b4 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -551,7 +551,7 @@ definitions: "$ref": "#/definitions/DynamicStreamCheckConfig" config_overrides: title: Config Overrides - description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated. + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. type: object additionalProperties: true examples: @@ -597,7 +597,7 @@ definitions: default: true config_overrides: title: Config Overrides - description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated. + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. type: object additionalProperties: true examples: diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index b20cdb4e4..803e820c1 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -79,7 +79,7 @@ class CheckDynamicStream(BaseModel): ) config_overrides: Optional[Dict[str, Any]] = Field( None, - description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated.", + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed.", examples=[{"max_waiting_time": 0}, {"page_size": 1}], title="Config Overrides", ) @@ -1789,7 +1789,7 @@ class CheckStream(BaseModel): dynamic_streams_check_configs: Optional[List[DynamicStreamCheckConfig]] = None config_overrides: Optional[Dict[str, Any]] = Field( None, - description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is and are not interpolated.", + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed.", examples=[{"max_waiting_time": 0}, {"page_size": 1}], title="Config Overrides", ) diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 9a488de5a..bbc978f31 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -1090,3 +1090,102 @@ def test_given_config_overrides_when_check_then_config_validations_run_against_t # `settings.mode` is overridden to a value outside the validator's enum, so this would fail were # the overlay validated instead of the config the user supplied. assert source.check(logger, config).status == Status.SUCCEEDED + + +def test_given_config_overrides_when_check_then_values_are_not_interpolated(): + """Pins the verbatim contract. A value containing `{{ }}` reaches components as that literal string, + so turning interpolation on later is a deliberate, test-breaking decision rather than a silent + reinterpretation of overrides already written.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "{{config['resource']}}"}, + } + ) + + with HttpMocker() as http_mocker: + literal_request = HttpRequest(url="https://api.test.com/%7B%7Bconfig%5B'resource'%5D%7D%7D") + http_mocker.get(literal_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(literal_request, 1) + + +_MANIFEST_WITH_CONFIG_DRIVEN_DYNAMIC_STREAM = { + "version": "6.7.0", + "type": "DeclarativeSource", + "check": {"type": "CheckDynamicStream", "stream_count": 1}, + "streams": [], + "dynamic_streams": [ + { + "type": "DynamicDeclarativeStream", + "name": "dynamic_items", + "stream_template": { + "type": "DeclarativeStream", + "name": "", + "primary_key": [], + "schema_loader": { + "type": "InlineSchemaLoader", + "schema": { + "$schema": "http://json-schema.org/schema#", + "type": "object", + "properties": {"id": {"type": "integer"}}, + }, + }, + "retriever": { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url": "https://api.test.com/{{ config['resource'] }}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": []}, + }, + "paginator": {"type": "NoPagination"}, + }, + }, + "components_resolver": { + "type": "ConfigComponentsResolver", + "stream_config": { + "type": "StreamConfig", + "configs_pointer": ["custom_streams"], + }, + "components_mapping": [ + { + "type": "ComponentMappingDefinition", + "field_path": ["name"], + "value": "{{components_values['name']}}", + } + ], + }, + } + ], +} + + +def test_given_config_overrides_on_check_dynamic_stream_then_components_see_them(): + """The overlay is read from the raw check definition, so it is checker-agnostic. Without this test a + refactor moving the read into `create_check_stream` would silently drop `CheckDynamicStream`.""" + config = {"resource": "sync", "custom_streams": [{"name": "items"}]} + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_DYNAMIC_STREAM) + manifest["check"] = { + "type": "CheckDynamicStream", + "stream_count": 1, + "config_overrides": {"resource": "check-only"}, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=config, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, config).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(overridden_request, 1) From 9b1c62bf88b44ed521ed56b84c19dd57db26e2c8 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Wed, 19 Aug 2026 18:35:09 +0300 Subject: [PATCH 3/6] fix(low-code): address review feedback on config_overrides Refuses `config_overrides` on a manifest that declares a `refresh_token_updater`. That turns an `OAuthAuthenticator` into a `DeclarativeSingleUseRefreshTokenOauth2Authenticator`, which is handed the config the component tree was built with - during a check, the overlay - and on refresh emits that entire dict as a CONNECTOR_CONFIG control message for the platform to persist. A check-only override would therefore become the connection's saved config and apply to every later sync, and restoring `self._config` afterwards cannot recall a message already written to stdout. Threading the config through more carefully would not help: the hazard is inherent to handing an overridden config to something whose job is to write the config back. Until the emitter is fixed to emit the config it was given plus only the token fields it owns, refusing the combination is the honest answer. Refuses override keys prefixed with `__airbyte`. Those are the platform's channel into the config rather than connector config - `CheckStream` reads `__airbyte_check_stream_names` out of the very config this overlay writes to - and a manifest that wants to choose which streams a check tests already has `stream_names`. Logs the overridden keys at INFO, keys only, since an override may name a secret field. Warns when an override key is absent from the spec's `connection_specification.properties`: the overlay is the one part of the config nothing validates, so a typo is otherwise a silent no-op. Corrects the shallow-copy paragraph in the docstring, which had it backwards. Because the copy is shallow, every nested object is shared with the config the source was constructed with, so a write into a nested path such as `("credentials", "access_token")` writes through to the user's config and the restore does not undo it. Only a write to a top-level key is discarded. Notes in `create_check_stream` and `create_check_dynamic_stream` that `model.config_overrides` is deliberately unread there because the source applies it around the whole check operation, so nobody wires it in twice or deletes it as dead. Co-Authored-By: Claude Opus 5 (1M context) --- .../concurrent_declarative_source.py | 104 +++++++++++++- .../parsers/model_to_component_factory.py | 4 + .../declarative/checks/test_check_stream.py | 135 ++++++++++++++++++ 3 files changed, 237 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 5d67501e0..fbdaa4c57 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -642,17 +642,31 @@ def _config_overridden_for_check( normalised itself nor propagated to fields derived from it by a `ConfigTransformation`. - `config_validations` run against `self._config_for_validation`, so an override cannot fail a validation written for the config as supplied. - - `self._config` is shared state. Reassigning it is not thread safe, and a component that writes - back into the config during check - a `SingleUseRefreshTokenOauth2Authenticator` refreshing its - token, say - writes into the throwaway overlay, which the restore then discards. The control - message is still emitted so the platform persists the new token, but an in-process read - following a check would carry the stale one. Both are acceptable while this is scoped to - `check`, which is one command per process. + + Two consequences of the copy being shallow and of `self._config` being shared state: + - Rebinding `self._config` is not thread safe. Acceptable only because `check` is one command + per process. + - The copy is shallow, so every nested object is the *same* object as in the config the source + was constructed with. A component that writes into a nested path - `dpath.new(config, + ("credentials", "access_token"), ...)` - therefore writes through to that shared dict, and the + restore does not undo it, because the restore only rebinds the top level. Only a write to a + top-level key is discarded. + + The write-through above is why a manifest that declares a `refresh_token_updater` is rejected + outright rather than documented: see `_raise_if_config_is_persisted`. """ if not config_overrides: yield return + self._raise_on_reserved_override_keys(config_overrides) + self._raise_if_config_is_persisted(config_overrides) + self._warn_on_unknown_override_keys(config_overrides) + # Keys only. An override may name a secret field, so values must not reach the logs. + self.logger.info( + f"Overriding config keys for the check operation: {', '.join(sorted(config_overrides))}" + ) + unmodified_config = self._config self._config = {**self._config, **config_overrides} try: @@ -660,6 +674,84 @@ def _config_overridden_for_check( finally: self._config = unmodified_config + @staticmethod + def _raise_on_reserved_override_keys(config_overrides: Mapping[str, Any]) -> None: + """Refuse to overlay keys in the platform's reserved `__airbyte` namespace. + + These are not connector config, they are the platform's channel into it: `CheckStream` reads + `__airbyte_check_stream_names` out of the very config this overlay writes to, so an override there + would change which streams a check tests. A manifest wanting that has `stream_names` for it. The + prefix is refused wholesale so future internal keys are covered too. + """ + reserved = sorted(key for key in config_overrides if key.startswith("__airbyte")) + if reserved: + raise ValueError( + f"`config_overrides` may not set the reserved key(s) {reserved}. Keys prefixed with " + "`__airbyte` belong to the platform, not to the connector's spec." + ) + + def _raise_if_config_is_persisted(self, config_overrides: Mapping[str, Any]) -> None: + """Reject `config_overrides` on a manifest whose authenticator writes the config back. + + A `refresh_token_updater` turns an `OAuthAuthenticator` into a + `DeclarativeSingleUseRefreshTokenOauth2Authenticator`, which is handed the config the component + tree was built with - during a check, the overlay. When it refreshes, `_emit_control_message` + prints that *entire* dict as a `CONNECTOR_CONFIG` control message, and the platform persists it. + So a check-only override would become the connection's saved config and apply to every later sync. + + Restoring `self._config` afterwards cannot help: the control message is already on stdout. And no + amount of threading the config through properly would help either, because the hazard is inherent + to handing an overridden config to something whose job is to write the config back. Until the + emitter is fixed to emit the config it was given plus only the token fields it owns, refusing the + combination is the honest answer. + """ + if not self._manifest_writes_back_config(self._source_config): + return + raise ValueError( + "`config_overrides` cannot be used by a manifest that declares a `refresh_token_updater`. " + "A token refresh during `check` emits the whole config it was handed as a CONNECTOR_CONFIG " + "control message, which the platform persists, so the check-only " + f"override(s) {sorted(config_overrides)} would be saved as this connection's config and " + "applied to every later sync. Remove `config_overrides`, or drop the `refresh_token_updater`." + ) + + @staticmethod + def _manifest_writes_back_config(definition: Any) -> bool: + """Whether any component in the manifest emits the connector config back to the platform.""" + if isinstance(definition, Mapping): + if definition.get("refresh_token_updater"): + return True + return any( + ConcurrentDeclarativeSource._manifest_writes_back_config(value) + for value in definition.values() + ) + if isinstance(definition, list): + return any( + ConcurrentDeclarativeSource._manifest_writes_back_config(item) + for item in definition + ) + return False + + def _warn_on_unknown_override_keys(self, config_overrides: Mapping[str, Any]) -> None: + """Warn about override keys the spec does not declare. + + The overlay is the one part of the config nothing validates - the entrypoint validates what the + user supplied, and `config_validations` deliberately run against `self._config_for_validation`. So + a typo or a since-renamed field is a silent no-op that surfaces much later as "why does check no + longer fail fast". + """ + if not self._spec_component: + return + declared = self._spec_component.connection_specification.get("properties") + if not isinstance(declared, Mapping): + return + unknown = sorted(key for key in config_overrides if key not in declared) + if unknown: + self.logger.warning( + f"Check-only config override(s) {unknown} are not declared in the connector spec, so " + "they will have no effect on any component that reads the config by field name." + ) + @property def dynamic_streams(self) -> List[Dict[str, Any]]: return self._dynamic_stream_configs( diff --git a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py index a96c406d8..1aa729d18 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -1306,6 +1306,9 @@ def create_check_stream( else [] ) + # `model.config_overrides` is deliberately not read here. The source applies it around the whole + # check operation (`ConcurrentDeclarativeSource._config_overridden_for_check`), which is what makes + # it work for every checker type rather than only this one. Do not wire it in a second time. return CheckStream( stream_names=model.stream_names or [], dynamic_streams_check_configs=dynamic_streams_check_configs, @@ -1320,6 +1323,7 @@ def create_check_dynamic_stream( use_check_availability = model.use_check_availability + # See `create_check_stream`: `model.config_overrides` is applied by the source, not here. return CheckDynamicStream( stream_count=model.stream_count, use_check_availability=use_check_availability, diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index bbc978f31..5c46c1bdd 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -1189,3 +1189,138 @@ def test_given_config_overrides_on_check_dynamic_stream_then_components_see_them assert source.check(logger, config).status == Status.SUCCEEDED http_mocker.assert_number_of_calls(overridden_request, 1) + + +_OAUTH_WITH_REFRESH_TOKEN_UPDATER = { + "type": "OAuthAuthenticator", + "token_refresh_endpoint": "https://api.test.com/oauth/token", + "client_id": "{{ config['credentials']['client_id'] }}", + "client_secret": "{{ config['credentials']['client_secret'] }}", + "refresh_token": "{{ config['credentials']['refresh_token'] }}", + "refresh_token_updater": {"type": "RefreshTokenUpdater", "refresh_token_name": "refresh_token"}, +} + + +def _manifest_with_refresh_token_updater(check_component): + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = check_component + manifest["streams"][0]["retriever"]["requester"]["authenticator"] = deepcopy( + _OAUTH_WITH_REFRESH_TOKEN_UPDATER + ) + return manifest + + +def test_given_refresh_token_updater_when_config_overrides_then_manifest_is_rejected(): + """A `refresh_token_updater` emits the whole config it was handed as a CONNECTOR_CONFIG control + message, which the platform persists - so a check-only override would become the connection's saved + config. The restore cannot recall a message already on stdout, so the combination is refused.""" + source = ConcurrentDeclarativeSource( + source_config=_manifest_with_refresh_token_updater( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ), + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with pytest.raises(ValueError, match="refresh_token_updater"): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + +def test_given_refresh_token_updater_without_config_overrides_then_nothing_is_rejected(): + """The rejection is scoped to the feature. A manifest that does not use `config_overrides` keeps + working with a `refresh_token_updater` exactly as before, even though the manifest scan detects it.""" + source = ConcurrentDeclarativeSource( + source_config=_manifest_with_refresh_token_updater( + {"type": "CheckStream", "stream_names": ["items"]} + ), + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source.resolved_manifest) is True + + # No overrides means no overlay, so the guard must not fire and the config must not be copied. + with source._config_overridden_for_check(None): + assert source._config is _CONFIG_DRIVEN_PATH_CONFIG + + +def test_given_config_overrides_when_check_then_overridden_keys_are_logged_without_values(caplog): + """An override may name a secret field, so the log records which keys were overridden but never what + they were set to.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "s3cr3t-value"}, + } + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/s3cr3t-value"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + with caplog.at_level(logging.INFO): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + assert "Overriding config keys for the check operation: resource" in caplog.text + assert "s3cr3t-value" not in caplog.text + + +def test_given_override_key_absent_from_the_spec_then_a_warning_is_logged(caplog): + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only", "typoed_key": 1}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": {"resource": {"type": "string"}}, + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/check-only"), + HttpResponse(body=json.dumps([{"id": 1}])), + ) + + with caplog.at_level(logging.WARNING): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + assert "typoed_key" in caplog.text + assert "not declared in the connector spec" in caplog.text + assert "'resource'" not in caplog.text + + +def test_given_an_airbyte_reserved_override_key_then_the_manifest_is_rejected(): + """`__airbyte`-prefixed keys are the platform's channel into the config - `CheckStream` reads + `__airbyte_check_stream_names` from the very config the overlay writes to. The feature refuses to + touch that namespace rather than leaving it available.""" + source = _source_with_check_component( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"__airbyte_check_stream_names": ["something_else"]}, + } + ) + + with pytest.raises(ValueError, match="__airbyte_check_stream_names"): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) From ae804ed56d270040d11f364b698f789a9e68cb38 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Thu, 20 Aug 2026 13:43:53 +0300 Subject: [PATCH 4/6] fix(low-code): detect refresh_token_updater taking all defaults `_manifest_writes_back_config` tested `refresh_token_updater` for truthiness. Every field of `RefreshTokenUpdater` has a default and none are required, so `refresh_token_updater: {}` is a valid way to take all of them - and it builds the same `DeclarativeSingleUseRefreshTokenOauth2Authenticator` a populated one does, because the factory's `if model.refresh_token_updater:` tests a model instance, which is always truthy. The transformer injects no `type` into it either, so the empty mapping stayed empty and slipped past the guard, leaving the config-persistence hazard open on exactly the shape the review reproduced. Testing `is not None` matches the factory on all three shapes: `{}`, populated, and absent. The rejection test is parametrized over the first two, and a new test covers the same OAuth authenticator without the updater, so the scan is pinned as not rejecting every OAuth manifest. Also document both rejections in the `config_overrides` description on `CheckStream` and `CheckDynamicStream`, since neither the `__airbyte` prefix restriction nor the `refresh_token_updater` one was discoverable before running a check. Co-Authored-By: Claude Opus 5 (1M context) --- .../concurrent_declarative_source.py | 11 +++- .../declarative_component_schema.yaml | 4 +- .../models/declarative_component_schema.py | 4 +- .../declarative/checks/test_check_stream.py | 50 ++++++++++++++++--- 4 files changed, 57 insertions(+), 12 deletions(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index fbdaa4c57..bbb54e4b0 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -717,9 +717,16 @@ def _raise_if_config_is_persisted(self, config_overrides: Mapping[str, Any]) -> @staticmethod def _manifest_writes_back_config(definition: Any) -> bool: - """Whether any component in the manifest emits the connector config back to the platform.""" + """Whether any component in the manifest emits the connector config back to the platform. + + Tested with `is not None` rather than for truthiness, to match the factory. Every field of + `RefreshTokenUpdater` has a default, so `refresh_token_updater: {}` is a valid way to take all of + them - and it builds a single-use authenticator just like a populated one, because the factory's + `if model.refresh_token_updater:` sees a model instance, which is always truthy. A truthiness test + here would see an empty dict and let that manifest through. + """ if isinstance(definition, Mapping): - if definition.get("refresh_token_updater"): + if definition.get("refresh_token_updater") is not None: return True return any( ConcurrentDeclarativeSource._manifest_writes_back_config(value) diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index 7506806b4..1de34175d 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -551,7 +551,7 @@ definitions: "$ref": "#/definitions/DynamicStreamCheckConfig" config_overrides: title: Config Overrides - description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync. type: object additionalProperties: true examples: @@ -597,7 +597,7 @@ definitions: default: true config_overrides: title: Config Overrides - description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. + description: Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync. type: object additionalProperties: true examples: diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index 803e820c1..25b50bdd9 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -79,7 +79,7 @@ class CheckDynamicStream(BaseModel): ) config_overrides: Optional[Dict[str, Any]] = Field( None, - description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed.", + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.", examples=[{"max_waiting_time": 0}, {"page_size": 1}], title="Config Overrides", ) @@ -1789,7 +1789,7 @@ class CheckStream(BaseModel): dynamic_streams_check_configs: Optional[List[DynamicStreamCheckConfig]] = None config_overrides: Optional[Dict[str, Any]] = Field( None, - description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed.", + description="Values overlaid onto the connector config for the duration of the check operation only. Use this when a check must behave differently from a sync - for example a shorter rate limit wait budget, so that check fails fast with a clear message instead of sleeping until the quota resets. Keys should be fields declared in the connector's spec. Values are used as-is. They are not interpolated, they replace a nested object rather than deep-merging into it, and they are applied after config migrations and transformations have run, so a field derived from an overridden field is not recomputed. Two combinations are rejected outright. Keys prefixed with `__airbyte` belong to the platform rather than to the connector's spec. And a manifest that declares a `refresh_token_updater` cannot use this field at all, because a token refresh during check emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config and apply to every later sync.", examples=[{"max_waiting_time": 0}, {"page_size": 1}], title="Config Overrides", ) diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 5c46c1bdd..548fdf5d5 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -1201,16 +1201,31 @@ def test_given_config_overrides_on_check_dynamic_stream_then_components_see_them } -def _manifest_with_refresh_token_updater(check_component): +def _manifest_with_refresh_token_updater(check_component, refresh_token_updater=None): manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) manifest["check"] = check_component - manifest["streams"][0]["retriever"]["requester"]["authenticator"] = deepcopy( - _OAUTH_WITH_REFRESH_TOKEN_UPDATER - ) + authenticator = deepcopy(_OAUTH_WITH_REFRESH_TOKEN_UPDATER) + if refresh_token_updater is not None: + authenticator["refresh_token_updater"] = deepcopy(refresh_token_updater) + manifest["streams"][0]["retriever"]["requester"]["authenticator"] = authenticator return manifest -def test_given_refresh_token_updater_when_config_overrides_then_manifest_is_rejected(): +@pytest.mark.parametrize( + "refresh_token_updater", + [ + pytest.param( + {"type": "RefreshTokenUpdater", "refresh_token_name": "refresh_token"}, id="populated" + ), + # Every field of `RefreshTokenUpdater` has a default, so an empty mapping is a valid way to take + # all of them. It builds the same single-use authenticator a populated one does, and the + # transformer injects no `type` into it, so it stays falsy - a truthiness test would miss it. + pytest.param({}, id="empty-taking-all-defaults"), + ], +) +def test_given_refresh_token_updater_when_config_overrides_then_manifest_is_rejected( + refresh_token_updater, +): """A `refresh_token_updater` emits the whole config it was handed as a CONNECTOR_CONFIG control message, which the platform persists - so a check-only override would become the connection's saved config. The restore cannot recall a message already on stdout, so the combination is refused.""" @@ -1220,7 +1235,8 @@ def test_given_refresh_token_updater_when_config_overrides_then_manifest_is_reje "type": "CheckStream", "stream_names": ["items"], "config_overrides": {"resource": "check-only"}, - } + }, + refresh_token_updater=refresh_token_updater, ), config=_CONFIG_DRIVEN_PATH_CONFIG, catalog=None, @@ -1231,6 +1247,28 @@ def test_given_refresh_token_updater_when_config_overrides_then_manifest_is_reje source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) +def test_given_no_refresh_token_updater_when_config_overrides_then_manifest_is_accepted(): + """The scan must not reject every OAuth manifest. The same authenticator without the updater writes + nothing back, so the overlay is allowed.""" + manifest = _manifest_with_refresh_token_updater( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + del manifest["streams"][0]["retriever"]["requester"]["authenticator"]["refresh_token_updater"] + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + def test_given_refresh_token_updater_without_config_overrides_then_nothing_is_rejected(): """The rejection is scoped to the feature. A manifest that does not use `config_overrides` keeps working with a `refresh_token_updater` exactly as before, even though the manifest scan detects it.""" From 2639ce4f7d846ed55a6e672418decca949451841 Mon Sep 17 00:00:00 2001 From: Anatolii Yatsuk Date: Thu, 20 Aug 2026 14:01:02 +0300 Subject: [PATCH 5/6] fix(low-code): stop the config-persistence scan from matching config values The scan for `refresh_token_updater` walks the whole raw manifest rather than only recognised authenticators, and that coarseness is load-bearing: it runs before references are resolved, so an authenticator reached through a `$ref` is found only because the walk also visits `definitions`. The net was too wide in two places that hold config values rather than components. A connector whose spec declares a property named `refresh_token_updater` lost `config_overrides` entirely - every use refused, naming an authenticator feature the manifest never declared - and an override of a config field by that name was refused for the same reason. Neither key can contain an authenticator: `spec` exists only at the top level of a manifest, and `config_overrides` only on `CheckStream` and `CheckDynamicStream`, so skipping both subtrees removes the false positives without narrowing detection. Adds a regression test per false positive, plus one for a `refresh_token_updater` behind a `$ref`, so the property the coarse walk exists to provide is pinned against a future narrowing. Co-Authored-By: Claude Opus 5 (1M context) --- .../concurrent_declarative_source.py | 16 +++- .../declarative/checks/test_check_stream.py | 87 +++++++++++++++++++ 2 files changed, 102 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index bbb54e4b0..92886be3b 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -715,10 +715,23 @@ def _raise_if_config_is_persisted(self, config_overrides: Mapping[str, Any]) -> "applied to every later sync. Remove `config_overrides`, or drop the `refresh_token_updater`." ) + # Subtrees of a manifest that hold data rather than components, so a `refresh_token_updater` key + # found inside one is a name collision and not an authenticator. Neither key is declared on any + # component that could contain an authenticator: `spec` exists only at the top level of a manifest, + # and `config_overrides` only on `CheckStream` and `CheckDynamicStream`. + _NON_COMPONENT_MANIFEST_KEYS = frozenset({"spec", "config_overrides"}) + @staticmethod def _manifest_writes_back_config(definition: Any) -> bool: """Whether any component in the manifest emits the connector config back to the platform. + The scan is deliberately coarse - it looks for the key anywhere rather than only under a + recognised authenticator - because it runs on the raw manifest, before references are resolved. + An authenticator reached through a `$ref` is only found because the walk also visits + `definitions`, where it lives under its own key rather than under a requester. The two subtrees + in `_NON_COMPONENT_MANIFEST_KEYS` are the exception: they hold config values, so a matching key + there means a connector whose spec happens to declare a field by that name, not a token refresh. + Tested with `is not None` rather than for truthiness, to match the factory. Every field of `RefreshTokenUpdater` has a default, so `refresh_token_updater: {}` is a valid way to take all of them - and it builds a single-use authenticator just like a populated one, because the factory's @@ -730,7 +743,8 @@ def _manifest_writes_back_config(definition: Any) -> bool: return True return any( ConcurrentDeclarativeSource._manifest_writes_back_config(value) - for value in definition.values() + for key, value in definition.items() + if key not in ConcurrentDeclarativeSource._NON_COMPONENT_MANIFEST_KEYS ) if isinstance(definition, list): return any( diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 548fdf5d5..819e301c8 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -1269,6 +1269,93 @@ def test_given_no_refresh_token_updater_when_config_overrides_then_manifest_is_a assert source._manifest_writes_back_config(source._source_config) is False +def test_given_spec_property_named_refresh_token_updater_then_overrides_are_allowed(): + """The scan walks the raw manifest looking for the key anywhere, because an authenticator reached + through a `$ref` is only found under `definitions`. A connector whose spec happens to declare a + config field by that name must not be caught by that net - nothing in `spec` is a component.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + manifest["spec"] = { + "type": "Spec", + "connection_specification": { + "$schema": "http://json-schema.org/draft-07/schema#", + "type": "object", + "properties": { + "resource": {"type": "string"}, + "refresh_token_updater": {"type": "string"}, + }, + }, + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + with HttpMocker() as http_mocker: + overridden_request = HttpRequest(url="https://api.test.com/check-only") + http_mocker.get(overridden_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + + +def test_given_override_of_a_field_named_refresh_token_updater_then_it_is_allowed(): + """`config_overrides` holds config values, not components, so a key that collides with the + authenticator field name is just a config field and must not trip the guard.""" + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only", "refresh_token_updater": "check-only"}, + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + assert source._manifest_writes_back_config(source._source_config) is False + + +def test_given_refresh_token_updater_behind_a_ref_then_manifest_is_rejected(): + """The coarse walk is what makes a `$ref`-ed authenticator detectable at all: the raw manifest holds + only the reference, and the authenticator itself sits under `definitions`. Narrowing the walk must + not lose that.""" + manifest = _manifest_with_refresh_token_updater( + { + "type": "CheckStream", + "stream_names": ["items"], + "config_overrides": {"resource": "check-only"}, + } + ) + manifest["definitions"] = { + "authenticator": manifest["streams"][0]["retriever"]["requester"]["authenticator"] + } + manifest["streams"][0]["retriever"]["requester"]["authenticator"] = { + "$ref": "#/definitions/authenticator" + } + + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with pytest.raises(ValueError, match="refresh_token_updater"): + source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG) + + def test_given_refresh_token_updater_without_config_overrides_then_nothing_is_rejected(): """The rejection is scoped to the feature. A manifest that does not use `config_overrides` keeps working with a `refresh_token_updater` exactly as before, even though the manifest scan detects it.""" From ccf055795283a5b6a290420452178fa5efa460d0 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 20 Aug 2026 11:08:05 +0000 Subject: [PATCH 6/6] fix(low-code): preserve check config_overrides from parameter propagation Co-Authored-By: bot_apk --- .../concurrent_declarative_source.py | 29 ++++++++++++- .../declarative/checks/test_check_stream.py | 42 +++++++++++++++++++ 2 files changed, 70 insertions(+), 1 deletion(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 92886be3b..60c51e4cb 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -217,6 +217,8 @@ def __init__( AlwaysLogSliceLogger() if emit_connector_builder_messages else DebugSliceLogger() ) + # Populated by `_pre_process_manifest` from the check component, before parameter propagation. + self._check_config_overrides: Optional[Mapping[str, Any]] = None # resolve all components in the manifest self._source_config = self._pre_process_manifest(dict(source_config)) # validate resolved manifest against the declarative component schema @@ -280,6 +282,10 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: manifest = self._fix_source_type(manifest) # Resolve references in the manifest resolved_manifest = ManifestReferenceResolver().preprocess_manifest(manifest) + # The check component's `config_overrides` holds connector config values rather than + # components, so it is read here: references are resolved, so a `check` behind a `$ref` is + # found, but parameters have not been propagated yet, so the values are still pristine. + self._check_config_overrides = self._extract_check_config_overrides(resolved_manifest) # Propagate types and parameters throughout the manifest propagated_manifest = ManifestComponentTransformer().propagate_types_and_parameters( "", resolved_manifest, {} @@ -287,6 +293,27 @@ def _pre_process_manifest(self, manifest: Dict[str, Any]) -> Dict[str, Any]: return propagated_manifest + @staticmethod + def _extract_check_config_overrides( + resolved_manifest: Mapping[str, Any], + ) -> Optional[Mapping[str, Any]]: + """Take the check component's `config_overrides` before parameter propagation mangles it. + + `ManifestComponentTransformer` injects `$parameters` and the parameters themselves into every + nested mapping that carries a truthy `type` key, because everywhere else in a manifest such a + mapping is a component. `config_overrides` values come from the connector's own spec, where + `type` is an ordinary field name - so an override like `credentials: {type: oauth, ...}` would + otherwise reach the check with stray keys in it. + """ + check = resolved_manifest.get("check") + if not isinstance(check, Mapping): + return None + config_overrides = check.get("config_overrides") + if not isinstance(config_overrides, Mapping): + return None + # Deep copy so the propagation that follows, which mutates in place, cannot reach these values. + return deepcopy(dict(config_overrides)) + def _fix_source_type(self, manifest: Dict[str, Any]) -> Dict[str, Any]: """ Fix the source type in the manifest. This is necessary because the source type is not always set in the manifest. @@ -612,7 +639,7 @@ def check(self, logger: logging.Logger, config: Mapping[str, Any]) -> AirbyteCon f"Expected to generate a ConnectionChecker component, but received {connection_checker.__class__}" ) - with self._config_overridden_for_check(check.get("config_overrides")): + with self._config_overridden_for_check(self._check_config_overrides): check_succeeded, error = connection_checker.check_connection(self, logger, self._config) if not check_succeeded: return AirbyteConnectionStatus(status=Status.FAILED, message=repr(error)) diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 819e301c8..5085a024c 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -1112,6 +1112,48 @@ def test_given_config_overrides_when_check_then_values_are_not_interpolated(): http_mocker.assert_number_of_calls(literal_request, 1) +def test_given_object_valued_config_override_when_check_then_parameters_are_not_injected_into_it(): + """`config_overrides` values come from the connector's spec, where `type` is an ordinary field name. + + `ManifestComponentTransformer` injects the enclosing component's parameters, plus a `$parameters` + key, into every nested mapping carrying a truthy `type` key - everywhere else in a manifest such a + mapping is a component. So a check component that declares `$parameters` used to hand the checker a + `credentials` object with stray keys in it. The request echoes the override's keys, so the mocked + request matches only if the object reached the stream exactly as authored. + """ + manifest = deepcopy(_MANIFEST_WITH_CONFIG_DRIVEN_PATH) + manifest["streams"][0]["retriever"]["requester"]["request_parameters"] = { + "credential_keys": "{{ config['credentials'].keys() | list | join(',') }}" + } + manifest["check"] = { + "type": "CheckStream", + "stream_names": ["items"], + # A no-op on a check component - `create_check_stream` passes `parameters={}` - but it is what + # the propagation walks with. + "$parameters": {"injected": "parameter"}, + "config_overrides": { + "resource": "check-only", + "credentials": {"type": "oauth", "client_id": "client-id"}, + }, + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=_CONFIG_DRIVEN_PATH_CONFIG, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + pristine_request = HttpRequest( + url="https://api.test.com/check-only", + query_params={"credential_keys": "type,client_id"}, + ) + http_mocker.get(pristine_request, HttpResponse(body=json.dumps([{"id": 1}]))) + + assert source.check(logger, _CONFIG_DRIVEN_PATH_CONFIG).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(pristine_request, 1) + + _MANIFEST_WITH_CONFIG_DRIVEN_DYNAMIC_STREAM = { "version": "6.7.0", "type": "DeclarativeSource",