Skip to content

fix: advance bookmark for streams that return no records #15

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Closed
jchumtong wants to merge 1 commit into from
Closed

fix: advance bookmark for streams that return no records #15

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 16 additions & 0 deletions tap_branch/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -10,6 +10,7 @@
from typing import TYPE_CHECKING, Any, ClassVar

import requests
from singer_sdk.helpers._state import increment_state
from singer_sdk.streams import Stream

if TYPE_CHECKING:
Expand All @@ -35,6 +36,7 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]:
"""Yield records by iterating over each day in the sync range."""
start = self.get_starting_timestamp(context).replace(hour=0, minute=0, second=0, microsecond=0)
end = datetime.now().astimezone(start.tzinfo).replace(hour=0, minute=0, second=0, microsecond=0) - timedelta(days=1)
last_synced: datetime | None = None
while start <= end:
self.logger.info("Syncing %s for %s", self.name, start)
urls = self._get_export_urls(start)
Expand All @@ -44,8 +46,22 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]:
"Downloading CSV file %d/%d for %s on %s", i, total, self.name, start.date()
)
yield from self._parse_csv(url, start)
last_synced = start
start += timedelta(days=1)

# Advance the bookmark to the last fully-scanned date even when the
# window produced no records. The replication key is otherwise only set
# from emitted rows, so streams that return nothing (e.g. eo_impression)
# would keep an empty bookmark and re-scan from start_date every run.
if last_synced is not None:
increment_state(
self.get_context_state(context),
latest_record={self.replication_key: last_synced},
replication_key=self.replication_key,
is_sorted=self.is_sorted,
check_sorted=False,
)

def _fetch_export_urls_for_date(self, export_date: date) -> dict[str, list[str]]:
"""Fetch all event-type URLs for a given date, with tap-level caching."""
cache: dict[date, dict[str, list[str]]] = self._tap.__dict__.setdefault(
Expand Down