From 982b6cdd0057cba53d7dd2968a8812f1ea08c1a2 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 23:55:43 +0000 Subject: [PATCH 1/4] fix: Handle FinalStateCursor gracefully and detect final-state for retention check - In _is_cursor_older_than_retention_period, continue to next cursor on NotImplementedError instead of raising SystemError. This allows the incremental cursor to be tried when the full_refresh cursor (e.g. FinalStateCursor) does not implement get_cursor_datetime_from_state. - In create_state_delegating_stream, detect final state (NO_CURSOR_STATE_KEY) and skip retention check entirely, returning incremental stream. When full_refresh completed without incremental_sync, the produced final state means we should always transition to incremental. - Add test for final-state scenario with api_retention_period. Co-Authored-By: alfredo.garcia@airbyte.io --- .../parsers/model_to_component_factory.py | 16 +++---- .../test_state_delegating_stream.py | 45 +++++++++++++++++++ 2 files changed, 51 insertions(+), 10 deletions(-) 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 788681d41b..a358dd0e94 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -637,6 +637,7 @@ WeekClampingStrategy, Weekday, ) +from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.concurrent.cursor import ( ConcurrentCursor, Cursor, @@ -3580,6 +3581,9 @@ def create_state_delegating_stream( if not stream_state and not has_parent: return self._create_component_from_model(model.full_refresh_stream, config=config, **kwargs) # type: ignore[no-any-return] + if stream_state and stream_state.get(NO_CURSOR_STATE_KEY): + return self._create_component_from_model(model.incremental_stream, config=config, **kwargs) # type: ignore[no-any-return] + incremental_stream: DefaultStream = self._create_component_from_model( model.incremental_stream, config=config, **kwargs ) # type: ignore[assignment] @@ -3615,20 +3619,12 @@ def _is_cursor_older_than_retention_period( for cursor in cursors: if not hasattr(cursor, "get_cursor_datetime_from_state"): - raise SystemError( - f"Stream '{stream_name}' cursor type '{type(cursor).__name__}' does not have " - f"get_cursor_datetime_from_state method. Cursor age validation with " - f"api_retention_period is not supported for this cursor type." - ) + continue try: cursor_datetime = cursor.get_cursor_datetime_from_state(stream_state) except NotImplementedError: - raise SystemError( - f"Stream '{stream_name}' cursor type '{type(cursor).__name__}' does not implement " - f"get_cursor_datetime_from_state. Cursor age validation with " - f"api_retention_period is not supported for this cursor type." - ) + continue if cursor_datetime is not None: break diff --git a/unit_tests/sources/declarative/test_state_delegating_stream.py b/unit_tests/sources/declarative/test_state_delegating_stream.py index 1a7df9c07a..f4db61b72e 100644 --- a/unit_tests/sources/declarative/test_state_delegating_stream.py +++ b/unit_tests/sources/declarative/test_state_delegating_stream.py @@ -566,3 +566,48 @@ def test_cursor_age_validation_raises_error_for_unparseable_cursor(): with pytest.raises(ValueError, match="not-a-date"): source.discover(logger=MagicMock(), config=_CONFIG) + + +def _create_manifest_with_no_full_refresh_cursor(api_retention_period: str) -> dict: + """Create a manifest where full_refresh_stream has no incremental_sync (gets FinalStateCursor).""" + manifest = copy.deepcopy(_MANIFEST) + manifest["definitions"]["TestStream"]["api_retention_period"] = api_retention_period + del manifest["definitions"]["TestStream"]["full_refresh_stream"]["incremental_sync"] + return manifest + + +@freezegun.freeze_time("2024-07-15") +def test_final_state_cursor_skips_retention_check_and_uses_incremental(): + """When state is a final state from FinalStateCursor, skip retention check and use incremental.""" + manifest = _create_manifest_with_retention_period("P7D") + + with HttpMocker() as http_mocker: + http_mocker.get( + HttpRequest( + url="https://api.test.com/items_with_filtration?start=2024-07-01&end=2024-07-15" + ), + HttpResponse( + body=json.dumps( + [ + {"id": 1, "name": "item_1", "updated_at": "2024-07-14"}, + ] + ) + ), + ) + + state = [ + AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name="TestStream", namespace=None), + stream_state=AirbyteStateBlob(__ab_no_cursor_state_message=True), + ), + ) + ] + source = ConcurrentDeclarativeSource( + source_config=manifest, config=_CONFIG, catalog=None, state=state + ) + configured_catalog = create_configured_catalog(source, _CONFIG) + + records = get_records(source, _CONFIG, configured_catalog, state) + assert len(records) == 1 From a7ffeb86f7f4d101c13a4a41150d4f0852a934a7 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 23:56:59 +0000 Subject: [PATCH 2/4] chore: Remove unused _create_manifest_with_no_full_refresh_cursor helper Co-Authored-By: alfredo.garcia@airbyte.io --- .../sources/declarative/test_state_delegating_stream.py | 8 -------- 1 file changed, 8 deletions(-) diff --git a/unit_tests/sources/declarative/test_state_delegating_stream.py b/unit_tests/sources/declarative/test_state_delegating_stream.py index f4db61b72e..e8a5573b56 100644 --- a/unit_tests/sources/declarative/test_state_delegating_stream.py +++ b/unit_tests/sources/declarative/test_state_delegating_stream.py @@ -568,14 +568,6 @@ def test_cursor_age_validation_raises_error_for_unparseable_cursor(): source.discover(logger=MagicMock(), config=_CONFIG) -def _create_manifest_with_no_full_refresh_cursor(api_retention_period: str) -> dict: - """Create a manifest where full_refresh_stream has no incremental_sync (gets FinalStateCursor).""" - manifest = copy.deepcopy(_MANIFEST) - manifest["definitions"]["TestStream"]["api_retention_period"] = api_retention_period - del manifest["definitions"]["TestStream"]["full_refresh_stream"]["incremental_sync"] - return manifest - - @freezegun.freeze_time("2024-07-15") def test_final_state_cursor_skips_retention_check_and_uses_incremental(): """When state is a final state from FinalStateCursor, skip retention check and use incremental.""" From 1f51404cd2c56266a396f200674f3d955ec93ec7 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 23:57:43 +0000 Subject: [PATCH 3/4] style: Fix ruff format for new lines in create_state_delegating_stream Co-Authored-By: alfredo.garcia@airbyte.io --- .../declarative/parsers/model_to_component_factory.py | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) 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 a358dd0e94..1314f68c6c 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -3579,10 +3579,14 @@ def create_state_delegating_stream( has_parent = False if has_parent_state is None else has_parent_state if not stream_state and not has_parent: - return self._create_component_from_model(model.full_refresh_stream, config=config, **kwargs) # type: ignore[no-any-return] + return self._create_component_from_model( + model.full_refresh_stream, config=config, **kwargs + ) # type: ignore[no-any-return] if stream_state and stream_state.get(NO_CURSOR_STATE_KEY): - return self._create_component_from_model(model.incremental_stream, config=config, **kwargs) # type: ignore[no-any-return] + return self._create_component_from_model( + model.incremental_stream, config=config, **kwargs + ) # type: ignore[no-any-return] incremental_stream: DefaultStream = self._create_component_from_model( model.incremental_stream, config=config, **kwargs From b16aa9053b6b901350dd6a959540e577048a89b8 Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Wed, 18 Feb 2026 23:58:17 +0000 Subject: [PATCH 4/4] style: Fix import sorting for NO_CURSOR_STATE_KEY Co-Authored-By: alfredo.garcia@airbyte.io --- .../sources/declarative/parsers/model_to_component_factory.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) 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 1314f68c6c..8f7f9fc551 100644 --- a/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py +++ b/airbyte_cdk/sources/declarative/parsers/model_to_component_factory.py @@ -618,6 +618,7 @@ NoopMessageRepository, ) from airbyte_cdk.sources.message.repository import StateFilteringMessageRepository +from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.call_rate import ( APIBudget, FixedWindowCallRatePolicy, @@ -637,7 +638,6 @@ WeekClampingStrategy, Weekday, ) -from airbyte_cdk.sources.streams import NO_CURSOR_STATE_KEY from airbyte_cdk.sources.streams.concurrent.cursor import ( ConcurrentCursor, Cursor,