Skip to content
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

Jump to bottom

Update replication_key state for descending streams #152

Closed
wants to merge 7 commits into from
Closed

Update replication_key state for descending streams #152

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
f274237
Properly update state signpost for desc streams
ericboucher Jun 27, 2022
d276513
Merge branch 'main' into desc-increment-state
ericboucher Jul 1, 2022
7caee70
Add test with mock API response
laurentS Jul 19, 2022
3dde9c1
Fix typing error
laurentS Jul 19, 2022
6d8a6fd
Merge branch 'main' into desc-increment-state
laurentS Nov 9, 2022
cb000fc
Update poetry.lock
laurentS Nov 9, 2022
8066fa8
Merge branch 'main' into desc-increment-state
ericboucher Nov 11, 2022
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
26 changes: 26 additions & 0 deletions tap_github/client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -54,6 +54,32 @@ def http_headers(self) -> Dict[str, str]:
headers["User-Agent"] = cast(str, self.config.get("user_agent", "tap-github"))
return headers

def get_records(self, context: Optional[dict]) -> Iterable[Dict[str, Any]]:
"""Return a generator of row-type dictionary objects.

Each row emitted should be a dictionary of property names to their values.

Args:
context: Stream partition or context dictionary.

Yields:
One item per (possibly processed) record in the API.
"""
yield from super().get_records(context)

# Important - Update state for streams in descending order
if self.use_fake_since_parameter:
state = self.get_context_state(context)
if set(["replication_key_signpost", "replication_key"]).issubset(
state.keys()
):
record: Dict = {}
record[state["replication_key"]] = state["replication_key_signpost"]
self._increment_stream_state(
latest_record=record,
context=context,
Comment on lines +71 to +83
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@ericboucher this will not run until get_records is actually iterated over, given the lazy nature of generators.

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

You mean at the end of super.get_records()? If so that's the goal yes. Or am I missing something? And get_records() needs to be called in a special way? Should this live somewhere else then? @edgarrmondragon

)

def get_next_page_token(
self, response: requests.Response, previous_token: Optional[Any]
) -> Optional[Any]:
Expand Down
2 changes: 2 additions & 0 deletions tap_github/repository_streams.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1001,6 +1001,8 @@ class CommitsStream(GitHubRestStream):
parent_stream_type = RepositoryStream
state_partitioning_keys = ["repo", "org"]
ignore_parent_replication_key = True
# Warning: /commits endpoint accept "since" but results are ordered by descending commit_timestamp
use_fake_since_parameter = True

def post_process(self, row: dict, context: Optional[Dict] = None) -> dict:
"""
Expand Down