From 91baf5313eb27f554f3ff6ad80463f0aa39fe610 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Thu, 22 Oct 2026 17:11:10 +0000 Subject: [PATCH 1/2] perf(low-code): skip dynamic streams when check reads only static streams Co-Authored-By: bot_apk --- .../concurrent_declarative_source.py | 7 +- .../declarative/checks/test_check_stream.py | 89 ++++++++++++++++++- 2 files changed, 90 insertions(+), 6 deletions(-) diff --git a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py index 7e7e6e387..c359bfcdb 100644 --- a/airbyte_cdk/sources/declarative/concurrent_declarative_source.py +++ b/airbyte_cdk/sources/declarative/concurrent_declarative_source.py @@ -491,7 +491,7 @@ def _uses_default_streams(self) -> bool: return type(self).streams is ConcurrentDeclarativeSource.streams def _streams_by_name(self, stream_names: List[str]) -> Optional[List[AbstractStream]]: - """Build the named static streams plus every dynamic stream. + """Build the named static streams without resolving or building dynamic streams. Return None when the names are empty, any name isn't static, or `streams()` is overridden. """ @@ -506,14 +506,13 @@ def _streams_by_name(self, stream_names: List[str]) -> Optional[List[AbstractStr return None self._prepare_for_stream_build() - dynamic_configs = self.dynamic_streams selected_names = set(stream_names) selected_configs = [ stream_config for stream_config in static_configs if stream_config.get("name") in selected_names - ] + dynamic_configs - return self._create_streams(selected_configs, static_configs + dynamic_configs) + ] + return self._create_streams(selected_configs, static_configs) def _first_streams(self, count: int) -> Optional[List[AbstractStream]]: """Build the first `count` streams, static then dynamic, in `streams()` order. diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index a05de813d..5628e7206 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -20,6 +20,7 @@ from airbyte_cdk.sources.streams.core import Stream from airbyte_cdk.sources.streams.http import HttpStream from airbyte_cdk.test.mock_http import HttpMocker, HttpRequest, HttpResponse +from airbyte_cdk.utils import AirbyteTracedException logger = logging.getLogger("test") config = dict() @@ -1113,6 +1114,7 @@ def test_check_for_dynamic_stream_name_falls_back_to_streams(mocker): state=None, ) streams_spy = mocker.spy(source, "streams") + dynamic_configs_spy = mocker.spy(source, "_dynamic_stream_configs") with HttpMocker() as http_mocker: http_mocker.get( @@ -1126,9 +1128,92 @@ def test_check_for_dynamic_stream_name_falls_back_to_streams(mocker): assert source.check(logger, source._config).status == Status.SUCCEEDED streams_spy.assert_called_once() + dynamic_configs_spy.assert_called_once() + + +def test_check_static_stream_does_not_resolve_http_dynamic_stream(): + manifest = deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT) + manifest["dynamic_streams"] = [manifest["dynamic_streams"][0]] + manifest["check"] = {"type": "CheckStream", "stream_names": ["static_stream"]} + source = ConcurrentDeclarativeSource( + source_config=manifest, + config={**_CONFIG, "api_key": "test"}, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/static"), + HttpResponse(body=json.dumps([])), + ) + assert source.check(logger, source._config).status == Status.SUCCEEDED + + +def test_check_with_duplicate_dynamic_streams_skips_resolution(): + manifest = deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT) + manifest["dynamic_streams"] = [manifest["dynamic_streams"][1]] + manifest["check"] = {"type": "CheckStream", "stream_names": ["static_stream"]} + config_with_duplicate_dynamic_streams = { + **_CONFIG, + "api_key": "test", + "custom_streams": [ + {"id": 1, "name": "duplicate"}, + {"id": 2, "name": "duplicate"}, + ], + } + source = ConcurrentDeclarativeSource( + source_config=manifest, + config=config_with_duplicate_dynamic_streams, + catalog=None, + state=None, + ) + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/static"), + HttpResponse(body=json.dumps([])), + ) + assert source.check(logger, source._config).status == Status.SUCCEEDED + + with pytest.raises( + AirbyteTracedException, + match="Dynamic streams list contains a duplicate name: duplicate. Please check your configuration.", + ): + source.discover(logger=source.logger, config=source._config) -def test_check_dynamic_stream_configs_are_evaluated_once_when_not_checked(mocker): +def test_check_skips_http_dynamic_stream_resolver_errors(): + manifest = deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT) + manifest["dynamic_streams"] = [manifest["dynamic_streams"][0]] + manifest["check"] = {"type": "CheckStream", "stream_names": ["static_stream"]} + source = ConcurrentDeclarativeSource( + source_config=manifest, + config={**_CONFIG, "api_key": "test"}, + catalog=None, + state=None, + ) + resolver_request = HttpRequest(url="https://api.test.com/items") + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/static"), + HttpResponse(body=json.dumps([])), + ) + http_mocker.get( + resolver_request, + HttpResponse(body=json.dumps({"error": "server error"}), status_code=500), + ) + assert source.check(logger, source._config).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(resolver_request, 0) + + with pytest.raises(AirbyteTracedException, match="HTTP Status Code: 500"): + source.discover(logger=source.logger, config=source._config) + + +def test_check_dynamic_stream_configs_are_not_evaluated_when_only_static_streams_are_checked( + mocker, +): manifest = deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT) manifest["dynamic_streams"] = [manifest["dynamic_streams"][1]] manifest["check"] = {"type": "CheckStream", "stream_names": ["static_stream"]} @@ -1147,7 +1232,7 @@ def test_check_dynamic_stream_configs_are_evaluated_once_when_not_checked(mocker dynamic_configs_spy = mocker.spy(source, "_dynamic_stream_configs") assert source.check(logger, source._config).status == Status.SUCCEEDED - dynamic_configs_spy.assert_called_once() + dynamic_configs_spy.assert_not_called() def test_check_builds_a_named_state_delegating_stream(mocker): From 2a09438a784f1f1eb0b6ded2f41360234753190d Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Fri, 23 Oct 2026 00:10:09 +0000 Subject: [PATCH 2/2] test(low-code): cover D2 stream resolution edge cases Co-Authored-By: bot_apk --- .../declarative/checks/test_check_stream.py | 44 +++++++++++++++++-- .../test_concurrent_declarative_source.py | 23 ++++++---- 2 files changed, 55 insertions(+), 12 deletions(-) diff --git a/unit_tests/sources/declarative/checks/test_check_stream.py b/unit_tests/sources/declarative/checks/test_check_stream.py index 5628e7206..9359f0f8c 100644 --- a/unit_tests/sources/declarative/checks/test_check_stream.py +++ b/unit_tests/sources/declarative/checks/test_check_stream.py @@ -1235,6 +1235,42 @@ def test_check_dynamic_stream_configs_are_not_evaluated_when_only_static_streams dynamic_configs_spy.assert_not_called() +def test_check_with_streams_override_resolves_http_dynamic_streams(): + class StreamsOverrideSource(ConcurrentDeclarativeSource): + def __init__(self, *args, **kwargs): + self.streams_call_count = 0 + super().__init__(*args, **kwargs) + + def streams(self, config): + self.streams_call_count += 1 + return super().streams(config=config) + + manifest = deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT) + manifest["dynamic_streams"] = [manifest["dynamic_streams"][0]] + manifest["check"] = {"type": "CheckStream", "stream_names": ["static_stream"]} + source = StreamsOverrideSource( + source_config=manifest, + config={**_CONFIG, "api_key": "test"}, + catalog=None, + state=None, + ) + resolver_request = HttpRequest(url="https://api.test.com/items") + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest(url="https://api.test.com/static"), + HttpResponse(body=json.dumps([])), + ) + http_mocker.get( + resolver_request, + HttpResponse(body=json.dumps([{"id": 3, "name": "item_3"}])), + ) + assert source.check(logger, source._config).status == Status.SUCCEEDED + http_mocker.assert_number_of_calls(resolver_request, 1) + + assert source.streams_call_count == 1 + + def test_check_builds_a_named_state_delegating_stream(mocker): stream = _stream_config("state_stream") incremental_stream = deepcopy(stream) @@ -1415,7 +1451,7 @@ def test_streams_by_name_enables_cache_for_checked_parent_without_mutating_manif assert source.resolved_manifest == manifest_before -def test_check_still_fails_when_a_dynamic_stream_cannot_be_built(mocker): +def test_static_check_succeeds_when_a_dynamic_stream_cannot_be_built(mocker): manifest = deepcopy(_MANIFEST_WITHOUT_CHECK_COMPONENT) manifest["dynamic_streams"] = [manifest["dynamic_streams"][1]] manifest["check"] = {"type": "CheckStream", "stream_names": ["static_stream"]} @@ -1446,8 +1482,10 @@ def create_component(*args, **kwargs): ) status = source.check(logger, source._config) - assert status.status == Status.FAILED - assert "Encountered an error while discovering streams" in status.message + assert status.status == Status.SUCCEEDED + + with pytest.raises(ValueError, match="bad dynamic stream"): + source.discover(logger=source.logger, config=source._config) def _stream_group_manifest(streams: list[dict], check_stream_name: str) -> dict: diff --git a/unit_tests/sources/declarative/test_concurrent_declarative_source.py b/unit_tests/sources/declarative/test_concurrent_declarative_source.py index 327bfa0e3..e5aa6a36d 100644 --- a/unit_tests/sources/declarative/test_concurrent_declarative_source.py +++ b/unit_tests/sources/declarative/test_concurrent_declarative_source.py @@ -6650,14 +6650,16 @@ def test_get_partition_router(stream_factory, expected_type): @pytest.mark.parametrize( - "build_path", + "build_path, expect_dynamic_stream_configs", [ - pytest.param("streams", id="streams"), - pytest.param("_streams_by_name", id="streams_by_name"), - pytest.param("_first_streams", id="first_streams"), + pytest.param("streams", True, id="streams"), + pytest.param("_streams_by_name", False, id="streams_by_name"), + pytest.param("_first_streams", True, id="first_streams"), ], ) -def test_api_budget_is_set_before_dynamic_streams_evaluated(build_path): +def test_api_budget_is_set_before_dynamic_streams_evaluated( + build_path, expect_dynamic_stream_configs +): """Verify all stream-building paths set the API budget before dynamic discovery.""" source = ConcurrentDeclarativeSource( source_config=_MANIFEST, config=_CONFIG, catalog=None, state=None @@ -6702,10 +6704,13 @@ def tracking_dynamic_stream_configs(*args, **kwargs): source._first_streams(1) assert "set_api_budget" in call_order, "set_api_budget was never called" - assert "dynamic_stream_configs" in call_order, "dynamic_stream_configs was never called" - assert call_order.index("set_api_budget") < call_order.index("dynamic_stream_configs"), ( - f"set_api_budget must be called before dynamic_stream_configs, but call order was: {call_order}" - ) + if expect_dynamic_stream_configs: + assert "dynamic_stream_configs" in call_order, "dynamic_stream_configs was never called" + assert call_order.index("set_api_budget") < call_order.index("dynamic_stream_configs"), ( + f"set_api_budget must be called before dynamic_stream_configs, but call order was: {call_order}" + ) + else: + assert "dynamic_stream_configs" not in call_order @pytest.mark.parametrize("count", [0, -1])