diff --git a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml index a02be6fa0e..30e86a739a 100644 --- a/airbyte_cdk/sources/declarative/declarative_component_schema.yaml +++ b/airbyte_cdk/sources/declarative/declarative_component_schema.yaml @@ -1357,11 +1357,11 @@ definitions: - "P1D" is_data_feed: title: Data Feed API - description: A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too. + description: A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value minus the lookback window, if one is set (never before the start date), and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too. type: boolean is_client_side_incremental: title: Client-side Incremental Filtering - description: Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window. + description: Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value minus the lookback window, if one is set (never before the start date), and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window. type: boolean is_compare_strictly: title: Strict Start-End Time Comparison diff --git a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py index a62c3db2fe..b698b7b255 100644 --- a/airbyte_cdk/sources/declarative/models/declarative_component_schema.py +++ b/airbyte_cdk/sources/declarative/models/declarative_component_schema.py @@ -1967,12 +1967,12 @@ class DatetimeBasedCursor(BaseModel): ) is_data_feed: Optional[bool] = Field( None, - description="A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too.", + description="A data feed API is an API that does not allow filtering and paginates the content from the most recent to the least recent. Given this, the CDK needs to know when to stop paginating and this field will generate a stop condition for pagination. The last page fetched still holds records that fall outside the cursor window, and those are filtered out as well, so Client-side Incremental Filtering does not need to be enabled alongside this field. Records are kept when their cursor value is within the window that starts at the previous sync's cursor value minus the lookback window, if one is set (never before the start date), and ends at the end date, defaulting to the current time, so records dated in the future are filtered out too.", title="Data Feed API", ) is_client_side_incremental: Optional[bool] = Field( None, - description="Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value (or the start date) and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window.", + description="Set to True if the target API endpoint does not take cursor values to filter records and returns all records anyway. This will cause the connector to filter out records locally, keeping only the ones whose cursor value falls within the window that starts at the previous sync's cursor value minus the lookback window, if one is set (never before the start date), and ends at the end date, defaulting to the current time. This means that all records would be read from the API, but only the records within that window will be emitted to the destination. This is not needed when Data Feed API is enabled, as a data feed already filters on the same window.", title="Client-side Incremental Filtering", ) is_compare_strictly: Optional[bool] = Field( diff --git a/airbyte_cdk/sources/streams/concurrent/cursor.py b/airbyte_cdk/sources/streams/concurrent/cursor.py index 11eaad235c..f603d1e27a 100644 --- a/airbyte_cdk/sources/streams/concurrent/cursor.py +++ b/airbyte_cdk/sources/streams/concurrent/cursor.py @@ -570,7 +570,18 @@ def should_be_synced(self, record: Record) -> bool: except ValueError: self._log_for_record_without_cursor_value() return True - return self.start <= record_cursor_value <= self._end_provider() + return self._lowest_cursor_value_to_sync() <= record_cursor_value <= self._end_provider() + + def _lowest_cursor_value_to_sync(self) -> CursorValueType: + # Keep the records the lookback window re-reads, never below the configured start. If the state is below the + # configured start, keep records from the state rather than raising the floor to the configured start. + if not self._lookback_window: + return self.start + try: + lower = self.start - self._lookback_window + except OverflowError: + return self.start + return max(lower, min(self.start, self._start)) if self._start else lower def _log_for_record_without_cursor_value(self) -> None: if not self._should_be_synced_logger_triggered: diff --git a/unit_tests/sources/declarative/incremental/test_client_side_incremental_lookback.py b/unit_tests/sources/declarative/incremental/test_client_side_incremental_lookback.py new file mode 100644 index 0000000000..97965f10a5 --- /dev/null +++ b/unit_tests/sources/declarative/incremental/test_client_side_incremental_lookback.py @@ -0,0 +1,253 @@ +# Copyright (c) 2026 Airbyte, Inc., all rights reserved. +from typing import Any, List, Mapping, MutableMapping, Optional + +import pytest +import requests_mock + +from airbyte_cdk.models import ( + AirbyteStateBlob, + AirbyteStateMessage, + AirbyteStateType, + AirbyteStreamState, + StreamDescriptor, +) +from airbyte_cdk.sources.declarative.concurrent_declarative_source import ( + ConcurrentDeclarativeSource, +) +from airbyte_cdk.test.catalog_builder import CatalogBuilder, ConfiguredAirbyteStreamBuilder +from airbyte_cdk.test.entrypoint_wrapper import EntrypointOutput, read + +_STATE_VALUE = 1714000000 +_DAY = 86400 +_START = _STATE_VALUE - 30 * _DAY +_CONFIG = {"start_date": str(_START)} +_URL_BASE = "https://api.example.com" +_STREAM_NAME = "conversation_parts" + +_CURSOR: Mapping[str, Any] = { + "type": "DatetimeBasedCursor", + "cursor_field": "updated_at", + "datetime_format": "%s", + "cursor_datetime_formats": ["%s"], + "start_datetime": { + "type": "MinMaxDatetime", + "datetime": "{{ config['start_date'] }}", + "datetime_format": "%s", + }, + "lookback_window": "P1D", + "is_client_side_incremental": True, +} + +_IN_LOOKBACK = {"id": "in_lookback", "updated_at": _STATE_VALUE - 3600} +_NEWER = {"id": "newer", "updated_at": _STATE_VALUE + 10} +_OLDER_THAN_LOOKBACK = {"id": "older_than_lookback", "updated_at": _STATE_VALUE - 2 * _DAY} +_BEFORE_START = {"id": "before_start", "updated_at": _START - 3600} + + +def _retriever(path: str, field_path: str) -> MutableMapping[str, Any]: + return { + "type": "SimpleRetriever", + "requester": { + "type": "HttpRequester", + "url": f"{_URL_BASE}{path}", + "http_method": "GET", + }, + "record_selector": { + "type": "RecordSelector", + "extractor": {"type": "DpathExtractor", "field_path": [field_path]}, + }, + } + + +def _manifest(kind: str, incremental_parent: bool = False) -> Mapping[str, Any]: + child_cursor = dict(_CURSOR) + if kind == "global": + child_cursor["global_substream_cursor"] = True + + if kind == "plain": + retriever = _retriever("/parts", "parts") + else: + parent: MutableMapping[str, Any] = { + "type": "DeclarativeStream", + "name": "conversations", + "primary_key": ["id"], + "schema_loader": {"type": "InlineSchemaLoader", "schema": {}}, + "retriever": _retriever("/conversations", "data"), + } + parent_config: MutableMapping[str, Any] = { + "type": "ParentStreamConfig", + "stream": parent, + "parent_key": "id", + "partition_field": "id", + } + if incremental_parent: + parent["incremental_sync"] = dict(_CURSOR) + parent_config["incremental_dependency"] = True + retriever = _retriever("/conversations/{{ stream_slice.id }}", "parts") + retriever["partition_router"] = { + "type": "SubstreamPartitionRouter", + "parent_stream_configs": [parent_config], + } + + return { + "version": "6.0.0", + "type": "DeclarativeSource", + "check": {"type": "CheckStream", "stream_names": [_STREAM_NAME]}, + "streams": [ + { + "type": "DeclarativeStream", + "name": _STREAM_NAME, + "primary_key": ["id"], + "schema_loader": {"type": "InlineSchemaLoader", "schema": {}}, + "retriever": retriever, + "incremental_sync": child_cursor, + } + ], + "spec": { + "type": "Spec", + "connection_specification": { + "type": "object", + "properties": {"start_date": {"type": "string"}}, + }, + }, + } + + +def _state(kind: str) -> MutableMapping[str, Any]: + if kind == "plain": + return {"updated_at": str(_STATE_VALUE)} + state: MutableMapping[str, Any] = { + "use_global_cursor": kind == "global", + "state": {"updated_at": str(_STATE_VALUE)}, + "lookback_window": 60, + } + if kind == "per_partition": + state["states"] = [ + { + "partition": {"id": 1, "parent_slice": {}}, + "cursor": {"updated_at": str(_STATE_VALUE)}, + } + ] + return state + + +def _read( + manifest: Mapping[str, Any], + state: Optional[Mapping[str, Any]], + responses: Mapping[str, Any], +) -> tuple[EntrypointOutput, List[str]]: + state_messages = ( + [ + AirbyteStateMessage( + type=AirbyteStateType.STREAM, + stream=AirbyteStreamState( + stream_descriptor=StreamDescriptor(name=_STREAM_NAME), + stream_state=AirbyteStateBlob(state), + ), + ) + ] + if state + else None + ) + catalog = ( + CatalogBuilder() + .with_stream(ConfiguredAirbyteStreamBuilder().with_name(_STREAM_NAME)) + .build() + ) + with requests_mock.Mocker() as m: + for path, body in responses.items(): + m.get(f"{_URL_BASE}{path}", json=body) + source = ConcurrentDeclarativeSource( + source_config=manifest, config=_CONFIG, catalog=catalog, state=state_messages + ) + output = read(source, _CONFIG, catalog, state_messages) + requested_paths = sorted({request.path for request in m.request_history}) + assert not output.errors, output.errors + return output, requested_paths + + +def _record_ids(output: EntrypointOutput) -> List[str]: + return sorted(message.record.data["id"] for message in output.records) + + +def _final_state(output: EntrypointOutput) -> Mapping[str, Any]: + return output.state_messages[-1].state.stream.stream_state.__dict__ + + +def _parts_responses(parts: List[Mapping[str, Any]]) -> Mapping[str, Any]: + return { + "/parts": {"parts": parts}, + "/conversations": {"data": [{"id": 1}]}, + "/conversations/1": {"parts": parts}, + } + + +@pytest.mark.parametrize("kind", ["plain", "per_partition", "global"]) +@pytest.mark.parametrize( + "with_newer_record, expected_state_value", + [(True, _STATE_VALUE + 10), (False, _STATE_VALUE)], + ids=["with_newer_record", "only_older_records"], +) +def test_given_state_and_lookback_window_when_read_then_emit_records_within_lookback( + kind: str, with_newer_record: bool, expected_state_value: int +) -> None: + parts = [_IN_LOOKBACK, _OLDER_THAN_LOOKBACK, _BEFORE_START] + ( + [_NEWER] if with_newer_record else [] + ) + + output, _ = _read(_manifest(kind), _state(kind), _parts_responses(parts)) + + assert _record_ids(output) == sorted(["in_lookback"] + (["newer"] if with_newer_record else [])) + final_state = _final_state(output) + cursor_state = final_state if kind == "plain" else final_state["state"] + assert int(cursor_state["updated_at"]) == expected_state_value + if kind == "per_partition": + assert int(final_state["states"][0]["cursor"]["updated_at"]) == expected_state_value + + +@pytest.mark.parametrize("kind", ["plain", "per_partition", "global"]) +def test_given_no_state_and_lookback_window_when_read_then_drop_records_before_start( + kind: str, +) -> None: + parts = [_IN_LOOKBACK, _NEWER, _OLDER_THAN_LOOKBACK, _BEFORE_START] + + output, _ = _read(_manifest(kind), None, _parts_responses(parts)) + + assert _record_ids(output) == ["in_lookback", "newer", "older_than_lookback"] + + +@pytest.mark.parametrize("kind", ["global", "per_partition"]) +@pytest.mark.parametrize( + "with_newer_parent", [True, False], ids=["with_newer_parent", "only_parent_in_lookback"] +) +def test_given_incremental_parent_with_lookback_when_read_then_parent_record_within_lookback_produces_child_records( + kind: str, with_newer_parent: bool +) -> None: + conversations = [{"id": 1, "updated_at": _STATE_VALUE - 3600}] + ( + [{"id": 2, "updated_at": _STATE_VALUE + 10}] if with_newer_parent else [] + ) + state = _state(kind) + state["parent_state"] = {"conversations": {"updated_at": str(_STATE_VALUE)}} + + output, requested_paths = _read( + _manifest(kind, incremental_parent=True), + state, + { + "/conversations": {"data": conversations}, + "/conversations/1": {"parts": [{"id": "p1", "updated_at": _STATE_VALUE - 3600}]}, + "/conversations/2": {"parts": [{"id": "p2", "updated_at": _STATE_VALUE + 10}]}, + }, + ) + + assert "/conversations/1" in requested_paths + assert "p1" in _record_ids(output) + final_state = _final_state(output) + assert int(final_state["state"]["updated_at"]) >= _STATE_VALUE + assert int(final_state["parent_state"]["conversations"]["updated_at"]) >= _STATE_VALUE + if kind == "per_partition": + partition_cursor = next( + partition_state["cursor"] + for partition_state in final_state["states"] + if partition_state["partition"]["id"] == 1 + ) + assert int(partition_cursor["updated_at"]) >= _STATE_VALUE diff --git a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py index d4b71b3ee1..862e247a4c 100644 --- a/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py +++ b/unit_tests/sources/declarative/retrievers/test_data_feed_integration.py @@ -34,7 +34,9 @@ def _manifest( - is_client_side_incremental: bool = False, partition_router: Optional[Mapping[str, Any]] = None + is_client_side_incremental: bool = False, + partition_router: Optional[Mapping[str, Any]] = None, + lookback_window: Optional[str] = None, ) -> Mapping[str, Any]: incremental_sync: dict[str, Any] = { "type": "DatetimeBasedCursor", @@ -49,6 +51,8 @@ def _manifest( } if is_client_side_incremental: incremental_sync["is_client_side_incremental"] = True + if lookback_window: + incremental_sync["lookback_window"] = lookback_window retriever: dict[str, Any] = { "type": "SimpleRetriever", @@ -208,6 +212,37 @@ def test_given_no_already_synced_records_then_paginate_until_the_end( assert record_ids == ["0", "1", "2", "3", "4"] +@pytest.mark.parametrize("is_client_side_incremental", [False, True]) +def test_given_lookback_window_then_paginate_through_the_lookback_and_emit_its_records( + is_client_side_incremental: bool, +) -> None: + pages_fetched, record_ids = _read( + _manifest(is_client_side_incremental=is_client_side_incremental, lookback_window="P1D"), + _state({"updated_at": "2021-01-01T00:00:00Z"}), + pages={ + # full pages, so that pagination is only stopped by the record older than the lookback window + "1": [ + {"id": "fresh_1", "updated_at": "2022-06-01T00:00:00Z"}, + {"id": "fresh_2", "updated_at": "2022-05-01T00:00:00Z"}, + {"id": "fresh_3", "updated_at": "2022-04-01T00:00:00Z"}, + {"id": "in_lookback", "updated_at": "2020-12-31T12:00:00Z"}, + ], + "2": [ + {"id": "before_lookback_1", "updated_at": "2020-12-30T00:00:00Z"}, + {"id": "before_lookback_2", "updated_at": "2020-12-29T00:00:00Z"}, + {"id": "before_lookback_3", "updated_at": "2020-12-28T00:00:00Z"}, + {"id": "before_lookback_4", "updated_at": "2020-12-27T00:00:00Z"}, + ], + "3": [ + {"id": "never_requested", "updated_at": "2020-12-26T00:00:00Z"}, + ], + }, + ) + + assert pages_fetched == ["1", "2"] + assert record_ids == ["fresh_1", "fresh_2", "fresh_3", "in_lookback"] + + def test_given_record_dated_in_the_future_then_filter_it_out() -> None: """ The retriever drops the records the cursor would not sync, and `should_be_synced` is bounded on both ends: with no diff --git a/unit_tests/sources/streams/concurrent/test_cursor.py b/unit_tests/sources/streams/concurrent/test_cursor.py index 13fe1df875..56d37e7c78 100644 --- a/unit_tests/sources/streams/concurrent/test_cursor.py +++ b/unit_tests/sources/streams/concurrent/test_cursor.py @@ -4,7 +4,7 @@ from copy import deepcopy from datetime import datetime, timedelta, timezone -from typing import Any, Mapping, Optional +from typing import Any, Mapping, MutableMapping, Optional from unittest import TestCase from unittest.mock import Mock @@ -34,6 +34,7 @@ from airbyte_cdk.sources.streams.concurrent.state_converters.datetime_stream_state_converter import ( CustomFormatConcurrentStreamStateConverter, EpochValueConcurrentStreamStateConverter, + IsoMillisConcurrentStreamStateConverter, ) from airbyte_cdk.sources.types import Record, StreamSlice @@ -1212,6 +1213,94 @@ def test_given_state_when_should_be_synced_then_use_cursor_value_to_filter(): ) +def _cursor_with_lookback(state: MutableMapping[str, Any]) -> ConcurrentCursor: + return ConcurrentCursor( + _A_STREAM_NAME, + _A_STREAM_NAMESPACE, + state, + Mock(spec=MessageRepository), + Mock(spec=ConnectorStateManager), + EpochValueConcurrentStreamStateConverter(True), + CursorField(_A_CURSOR_FIELD_KEY), + _SLICE_BOUNDARY_FIELDS, + datetime.fromtimestamp(_SHOULD_BE_SYNCED_START, timezone.utc), + EpochValueConcurrentStreamStateConverter.get_end_provider(), + timedelta(seconds=3), + ) + + +def _is_synced(cursor: ConcurrentCursor, cursor_value: int) -> bool: + return cursor.should_be_synced( + Record(data={_A_CURSOR_FIELD_KEY: cursor_value}, stream_name="test_stream") + ) + + +def test_given_state_and_lookback_window_when_should_be_synced_then_keep_records_within_lookback(): + state_value = _SHOULD_BE_SYNCED_START + 5 + cursor = _cursor_with_lookback({_A_CURSOR_FIELD_KEY: state_value}) + + assert _is_synced(cursor, state_value - 3) + assert not _is_synced(cursor, state_value - 4) + + +def test_given_lookback_window_when_should_be_synced_then_never_keep_records_before_start(): + cursor = _cursor_with_lookback({}) + assert _is_synced(cursor, _SHOULD_BE_SYNCED_START) + assert not _is_synced(cursor, _SHOULD_BE_SYNCED_START - 1) + + cursor = _cursor_with_lookback({_A_CURSOR_FIELD_KEY: _SHOULD_BE_SYNCED_START + 1}) + assert not _is_synced(cursor, _SHOULD_BE_SYNCED_START - 1) + + +def test_given_concurrent_state_before_start_and_lookback_window_when_should_be_synced_then_keep_records_from_state(): + state_value = _SHOULD_BE_SYNCED_START - 5 + cursor = _cursor_with_lookback( + {"state_type": "date-range", "slices": [{"start": 0, "end": state_value}]} + ) + + assert _is_synced(cursor, state_value + 2) + assert not _is_synced(cursor, state_value - 1) + + +def test_given_no_start_nor_state_and_lookback_window_when_should_be_synced_then_do_not_overflow(): + cursor = ConcurrentCursor( + _A_STREAM_NAME, + _A_STREAM_NAMESPACE, + {}, + Mock(spec=MessageRepository), + Mock(spec=ConnectorStateManager), + IsoMillisConcurrentStreamStateConverter(), + CursorField(_A_CURSOR_FIELD_KEY), + _SLICE_BOUNDARY_FIELDS, + None, + IsoMillisConcurrentStreamStateConverter.get_end_provider(), + timedelta(days=1), + ) + + assert cursor.should_be_synced( + Record(data={_A_CURSOR_FIELD_KEY: "2024-01-01T00:00:00.000Z"}, stream_name="test_stream") + ) + + +def test_given_state_without_start_and_lookback_window_when_should_be_synced_then_keep_records_within_lookback(): + cursor = ConcurrentCursor( + _A_STREAM_NAME, + _A_STREAM_NAMESPACE, + {_A_CURSOR_FIELD_KEY: 10}, + Mock(spec=MessageRepository), + Mock(spec=ConnectorStateManager), + EpochValueConcurrentStreamStateConverter(True), + CursorField(_A_CURSOR_FIELD_KEY), + _SLICE_BOUNDARY_FIELDS, + None, + EpochValueConcurrentStreamStateConverter.get_end_provider(), + timedelta(seconds=3), + ) + + assert _is_synced(cursor, 7) + assert not _is_synced(cursor, 6) + + def test_given_partitioned_state_without_slices_nor_start_when_should_be_synced_then_use_zero_value_to_filter(): cursor = ConcurrentCursor( _A_STREAM_NAME,