From 9d77434b02ed1b0cb848655790bb284d924a9e34 Mon Sep 17 00:00:00 2001 From: Atif Imam Date: Sun, 30 Aug 2026 18:41:42 +0530 Subject: [PATCH 1/6] fix: migrate from get_next_page_token to get_new_paginator Replace deprecated RESTStream.get_next_page_token with the new paginator API. Add GitHubRestPaginator and GitHubGraphQLPaginator in client.py and update the four repository streams that had custom early-exit pagination logic to use get_new_paginator with has_more overrides. Closes #171 --- tap_github/client.py | 298 ++++++++++++++++--------------- tap_github/repository_streams.py | 197 ++++++++++---------- 2 files changed, 250 insertions(+), 245 deletions(-) diff --git a/tap_github/client.py b/tap_github/client.py index cda1aece..9efbfb87 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -13,6 +13,7 @@ from nested_lookup import nested_lookup from singer_sdk.exceptions import FatalAPIError, RetriableAPIError from singer_sdk.helpers.jsonpath import extract_jsonpath +from singer_sdk.pagination import BaseAPIPaginator from singer_sdk.streams import GraphQLStream, RESTStream from tap_github.authenticator import GitHubTokenAuthenticator @@ -28,96 +29,41 @@ EMPTY_REPO_ERROR_STATUS = 409 -class GitHubRestStream(RESTStream): - """GitHub Rest stream class.""" - - MAX_PER_PAGE = 100 # GitHub's limit is 100. - MAX_RESULTS_LIMIT: int | None = None - DEFAULT_API_BASE_URL = "https://api.github.com" - LOG_REQUEST_METRIC_URLS = True - - # GitHub is missing the "since" parameter on a few endpoints - # set this parameter to True if your stream needs to navigate data in descending order # noqa: E501 - # and try to exit early on its own. - # This only has effect on streams whose `replication_key` is `updated_at`. - use_fake_since_parameter = False +class GitHubRestPaginator(BaseAPIPaginator[int | str | None]): + """Paginator for GitHub REST API streams.""" - # Set to True to use cursor-based pagination instead of page-based pagination - use_cursor_pagination = False - - _authenticator: GitHubTokenAuthenticator | None = None + def __init__(self, stream: GitHubRestStream, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 + super().__init__(None, *args, **kwargs) + self.stream = stream - @property - def authenticator(self) -> GitHubTokenAuthenticator: - if self._authenticator is None: - self._authenticator = GitHubTokenAuthenticator.from_stream(self) - return self._authenticator - - @property - def url_base(self) -> str: - return self.config.get("api_url_base", self.DEFAULT_API_BASE_URL) - - primary_keys: ClassVar[list[str]] = ["id"] - replication_key: str | None = None - tolerated_http_errors: ClassVar[list[int]] = [] - - @property - def http_headers(self) -> dict[str, str]: - """Return the http headers needed.""" - headers = {"Accept": "application/vnd.github.v3+json"} - headers["User-Agent"] = cast("str", self.config.get("user_agent", "tap-github")) - return headers - - def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]: - """ - Override parent method to set organization-specific authentication - before fetching records. - """ - # Set organization-specific authentication before fetching records - if context is not None and "org" in context: - self.authenticator.set_organization(context["org"]) - - yield from super().get_records(context) - - def get_next_page_token( - self, - response: requests.Response, - previous_token: Any | None, # noqa: ANN401 - ) -> Any | None: # noqa: ANN401 - """Return a token for identifying next page or None if no more pages.""" + def has_more(self, response: requests.Response) -> bool: + """Check if there are more pages.""" if ( - previous_token - and self.MAX_RESULTS_LIMIT - and not self.use_cursor_pagination + self.current_value + and self.stream.MAX_RESULTS_LIMIT + and not self.stream.use_cursor_pagination and ( - cast("int", previous_token) * self.MAX_PER_PAGE - >= self.MAX_RESULTS_LIMIT + cast("int", self.current_value) * self.stream.MAX_PER_PAGE + >= self.stream.MAX_RESULTS_LIMIT ) ): - return None + return False - # Leverage header links returned by the GitHub API. if "next" not in response.links: - return None + return False resp_json = response.json() results = ( resp_json if isinstance(resp_json, list) - else list(extract_jsonpath(self.records_jsonpath, input=resp_json)) + else list(extract_jsonpath(self.stream.records_jsonpath, input=resp_json)) ) - # Exit early if the response has no items. ? Maybe duplicative the "next" link check. # noqa: E501 if not results: - return None + return False - # Unfortunately endpoints such as /starred, /stargazers, /events and /pulls do not support # noqa: E501 - # the "since" parameter out of the box. So we use a workaround here to exit early. # noqa: E501 - # For such streams, we sort by descending dates (most recent first), and paginate # noqa: E501 - # "back in time" until we reach records before our "fake_since" parameter. - if self.replication_key and self.use_fake_since_parameter: + if self.stream.replication_key and self.stream.use_fake_since_parameter: request_parameters = parse_qs(str(urlparse(response.request.url).query)) - # parse_qs interprets "+" as a space, revert this to keep an aware datetime try: since = ( request_parameters["fake_since"][0].replace(" ", "+") @@ -125,7 +71,7 @@ def get_next_page_token( else "" ) except IndexError: - return None + return False direction = ( request_parameters["direction"][0] @@ -133,27 +79,27 @@ def get_next_page_token( else None ) - # commit_timestamp is a constructed key which does not exist in the raw response # noqa: E501 replication_date = ( - results[-1][self.replication_key] - if self.replication_key != "commit_timestamp" + results[-1][self.stream.replication_key] + if self.stream.replication_key != "commit_timestamp" else results[-1]["commit"]["committer"]["date"] ) - # exit early if the replication_date is before our since parameter if ( since and direction == "desc" and (parse(replication_date) < parse(since)) ): - return None + return False + + return True - # Handle cursor-based pagination - if self.use_cursor_pagination: + def get_next(self, response: requests.Response) -> int | str | None: + """Get the next pagination token.""" + if self.stream.use_cursor_pagination: parsed_url = urlparse(response.links["next"]["url"]) captured_after_value_list = parse_qs(parsed_url.query).get("after") return captured_after_value_list[0] if captured_after_value_list else None - # Use header links returned by the GitHub API for page-based pagination. parsed_url = urlparse(response.links["next"]["url"]) captured_page_value_list = parse_qs(parsed_url.query).get("page") next_page_string = ( @@ -162,7 +108,127 @@ def get_next_page_token( if next_page_string and next_page_string.isdigit(): return int(next_page_string) - return (previous_token or 1) + 1 + current = self.current_value + if isinstance(current, int): + return current + 1 + return 2 + + +class GitHubGraphQLPaginator(BaseAPIPaginator[dict[str, str] | None]): + """Paginator for GitHub GraphQL API streams.""" + + def __init__(self, stream: GitHubGraphqlStream, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 + super().__init__(None, *args, **kwargs) + self.stream = stream + + def has_more(self, response: requests.Response) -> bool: + """Check if there are more pages.""" + resp_json = response.json() + next_page_results = nested_lookup( + key="hasNextPage_", + document=resp_json, + wild=True, + with_keys=True, + ) + has_next_page_indices: list[int] = [] + for key, value in next_page_results.items(): + if any(value): + pagination_index = int(str(key).split("_")[1]) + has_next_page_indices.append(pagination_index) + return len(has_next_page_indices) > 0 + + def get_next(self, response: requests.Response) -> dict[str, str] | None: + """Get the next pagination token.""" + resp_json = response.json() + next_page_results = nested_lookup( + key="hasNextPage_", + document=resp_json, + wild=True, + with_keys=True, + ) + has_next_page_indices: list[int] = [] + for key, value in next_page_results.items(): + if any(value): + pagination_index = int(str(key).split("_")[1]) + has_next_page_indices.append(pagination_index) + + if not has_next_page_indices: + return None + + max_pagination_index = max(has_next_page_indices) + next_page_cursors: dict[str, str] = {} + for key, value in (self.current_value or {}).items(): + pagination_index = int(str(key).split("_")[1]) + if pagination_index < max_pagination_index: + next_page_cursors[key] = value + + next_page_end_cursor_results = nested_lookup( + key=f"endCursor_{max_pagination_index}", + document=resp_json, + ) + next_page_key = f"nextPageCursor_{max_pagination_index}" + next_page_cursor = next( + cursor for cursor in next_page_end_cursor_results if cursor is not None + ) + next_page_cursors[next_page_key] = next_page_cursor + + return next_page_cursors + + +class GitHubRestStream(RESTStream): + """GitHub Rest stream class.""" + + MAX_PER_PAGE = 100 # GitHub's limit is 100. + MAX_RESULTS_LIMIT: int | None = None + DEFAULT_API_BASE_URL = "https://api.github.com" + LOG_REQUEST_METRIC_URLS = True + + # GitHub is missing the "since" parameter on a few endpoints + # set this parameter to True if your stream needs to navigate data in descending order # noqa: E501 + # and try to exit early on its own. + # This only has effect on streams whose `replication_key` is `updated_at`. + use_fake_since_parameter = False + + # Set to True to use cursor-based pagination instead of page-based pagination + use_cursor_pagination = False + + _authenticator: GitHubTokenAuthenticator | None = None + + @property + def authenticator(self) -> GitHubTokenAuthenticator: + if self._authenticator is None: + self._authenticator = GitHubTokenAuthenticator.from_stream(self) + return self._authenticator + + @property + def url_base(self) -> str: + return self.config.get("api_url_base", self.DEFAULT_API_BASE_URL) + + primary_keys: ClassVar[list[str]] = ["id"] + replication_key: str | None = None + tolerated_http_errors: ClassVar[list[int]] = [] + + @property + def http_headers(self) -> dict[str, str]: + """Return the http headers needed.""" + headers = {"Accept": "application/vnd.github.v3+json"} + headers["User-Agent"] = cast("str", self.config.get("user_agent", "tap-github")) + return headers + + def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]: + """ + Override parent method to set organization-specific authentication + before fetching records. + """ + # Set organization-specific authentication before fetching records + if context is not None and "org" in context: + self.authenticator.set_organization(context["org"]) + + yield from super().get_records(context) + + def get_new_paginator(self) -> BaseAPIPaginator | None: + """Get a new paginator for this stream.""" + return GitHubRestPaginator(self) def get_url_params( self, @@ -437,71 +503,9 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]: if record is not None: yield record - def get_next_page_token( - self, - response: requests.Response, - previous_token: Any | None, # noqa: ANN401 - ) -> Any | None: # noqa: ANN401 - """ - Return a dict of cursors for identifying next page or None if no more pages. - - Note - pagination requires the Graphql query to have nextPageCursor_X parameters - with the assosciated hasNextPage_X, startCursor_X and endCursor_X. - - X should be an integer between 0 and 9, increasing with query depth. - - Warning - we recommend to avoid using deep (nested) pagination. - """ - - resp_json = response.json() - - # Find if results contains "hasNextPage_X" flags and if any are True. - # If so, set nextPageCursor_X to endCursor_X for X max. - - next_page_results = nested_lookup( - key="hasNextPage_", - document=resp_json, - wild=True, - with_keys=True, - ) - - has_next_page_indices: list[int] = [] - # Iterate over all the items and filter items with hasNextPage = True. - for key, value in next_page_results.items(): - # Check if key is even then add pair to new dictionary - if any(value): - pagination_index = int(str(key).split("_")[1]) - has_next_page_indices.append(pagination_index) - - # Check if any "hasNextPage" is True. Otherwise, exit early. - if not len(has_next_page_indices) > 0: - return None - - # Get deepest pagination item - max_pagination_index = max(has_next_page_indices) - - # We leverage previous_token to remember the pagination cursors - # for indices below max_pagination_index. - next_page_cursors: dict[str, str] = {} - for key, value in (previous_token or {}).items(): - # Only keep pagination info for indices below max_pagination_index. - pagination_index = int(str(key).split("_")[1]) - if pagination_index < max_pagination_index: - next_page_cursors[key] = value - - # Get the pagination cursor to update and increment it. - next_page_end_cursor_results = nested_lookup( - key=f"endCursor_{max_pagination_index}", - document=resp_json, - ) - - next_page_key = f"nextPageCursor_{max_pagination_index}" - next_page_cursor = next( - cursor for cursor in next_page_end_cursor_results if cursor is not None - ) - next_page_cursors[next_page_key] = next_page_cursor - - return next_page_cursors + def get_new_paginator(self) -> BaseAPIPaginator | None: + """Get a new paginator for this stream.""" + return GitHubGraphQLPaginator(self) def get_url_params( self, diff --git a/tap_github/repository_streams.py b/tap_github/repository_streams.py index 0be3e683..d8fb5a6f 100644 --- a/tap_github/repository_streams.py +++ b/tap_github/repository_streams.py @@ -12,7 +12,12 @@ from singer_sdk.exceptions import FatalAPIError, RetriableAPIError from singer_sdk.helpers.jsonpath import extract_jsonpath -from tap_github.client import GitHubDiffStream, GitHubGraphqlStream, GitHubRestStream +from tap_github.client import ( + GitHubDiffStream, + GitHubGraphQLPaginator, + GitHubGraphqlStream, + GitHubRestStream, +) from tap_github.schema_objects import ( files_object, label_object, @@ -2064,37 +2069,33 @@ def post_process(self, row: dict, context: Context | None = None) -> dict: row["user_id"] = row["user"]["id"] return row - def get_next_page_token( - self, - response: requests.Response, - previous_token: Any | None, # noqa: ANN401 - ) -> Any | None: # noqa: ANN401 - """ - Exit early if a since parameter is provided. - """ - request_parameters = parse_qs(str(urlparse(response.request.url).query)) + def get_new_paginator(self): # noqa: ANN201 + """Get a new paginator for this stream.""" + stream = self - # parse_qs interprets "+" as a space, revert this to keep an aware datetime - try: - since = ( - request_parameters["since"][0].replace(" ", "+") - if "since" in request_parameters - else "" - ) - except IndexError: - since = "" - - # If since parameter is present, try to exit early by looking at the last "starred_at". # noqa: E501 - # Noting that we are traversing in DESCENDING order by STARRED_AT. - if since: - results = list(extract_jsonpath(self.query_jsonpath, input=response.json())) - # If no results, return None to exit early. - if len(results) == 0: - return None - last = results[-1] - if parse(last["starred_at"]) < parse(since): - return None - return super().get_next_page_token(response, previous_token) + class StargazersPaginator(GitHubGraphQLPaginator): + def has_more(self, response) -> bool: # noqa: ANN001 + request_parameters = parse_qs(str(urlparse(response.request.url).query)) + try: + since = ( + request_parameters["since"][0].replace(" ", "+") + if "since" in request_parameters + else "" + ) + except IndexError: + since = "" + if since: + results = list( + extract_jsonpath(stream.query_jsonpath, input=response.json()) + ) + if len(results) == 0: + return False + last = results[-1] + if parse(last["starred_at"]) < parse(since): + return False + return super().has_more(response) + + return StargazersPaginator(stream) @property def query(self) -> str: @@ -2277,27 +2278,29 @@ def get_url_params( self.cutoff = self.get_starting_timestamp(context) return super().get_url_params(context, next_page_token) - def get_next_page_token( - self, - response: requests.Response, - previous_token: Any | None, # noqa: ANN401 - ) -> Any | None: # noqa: ANN401 - """ - Exit early if oldest updated_at is older than the replication bookmark. - """ - self.logger.debug("Cutoff: %s", self.cutoff) - if self.cutoff: - results = list(extract_jsonpath(self.query_jsonpath, input=response.json())) - if results: - oldest_updated_at = parse(results[-1][self.replication_key]) - if oldest_updated_at < self.cutoff: - self.logger.info( - "Early exit: oldest=%s, cutoff=%s", - oldest_updated_at, - self.cutoff, + def get_new_paginator(self): # noqa: ANN201 + """Get a new paginator for this stream.""" + stream = self + + class DiscussionsPaginator(GitHubGraphQLPaginator): + def has_more(self, response) -> bool: # noqa: ANN001 + stream.logger.debug("Cutoff: %s", stream.cutoff) + if stream.cutoff: + results = list( + extract_jsonpath(stream.query_jsonpath, input=response.json()) ) - return None # early exit - return super().get_next_page_token(response, previous_token) + if results: + oldest_updated_at = parse(results[-1][stream.replication_key]) + if oldest_updated_at < stream.cutoff: + stream.logger.info( + "Early exit: oldest=%s, cutoff=%s", + oldest_updated_at, + stream.cutoff, + ) + return False + return super().has_more(response) + + return DiscussionsPaginator(stream) def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """ @@ -2597,29 +2600,29 @@ def get_url_params( self.cutoff = self.get_starting_timestamp(context) return super().get_url_params(context, next_page_token) - def get_next_page_token( - self, - response: requests.Response, - previous_token: Any | None, # noqa: ANN401 - ) -> Any | None: # noqa: ANN401 - """ - Exit early if first (oldest) record in the page is older than the replication - bookmark. With github's default record ordering, each page contains records - in ascending order. - """ - self.logger.debug("Cutoff: %s", self.cutoff) - if self.cutoff: - results = list(extract_jsonpath(self.query_jsonpath, input=response.json())) - if results: - oldest_created_at = parse(results[0][self.replication_key]) - if oldest_created_at < self.cutoff: - self.logger.info( - "Early exit: oldest=%s, cutoff=%s", - oldest_created_at, - self.cutoff, + def get_new_paginator(self): # noqa: ANN201 + """Get a new paginator for this stream.""" + stream = self + + class DiscussionCommentsPaginator(GitHubGraphQLPaginator): + def has_more(self, response) -> bool: # noqa: ANN001 + stream.logger.debug("Cutoff: %s", stream.cutoff) + if stream.cutoff: + results = list( + extract_jsonpath(stream.query_jsonpath, input=response.json()) ) - return None # early exit - return super().get_next_page_token(response, previous_token) + if results: + oldest_created_at = parse(results[0][stream.replication_key]) + if oldest_created_at < stream.cutoff: + stream.logger.info( + "Early exit: oldest=%s, cutoff=%s", + oldest_created_at, + stream.cutoff, + ) + return False + return super().has_more(response) + + return DiscussionCommentsPaginator(stream) def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """ @@ -2858,32 +2861,30 @@ def get_url_params( self.cutoff = self.get_starting_timestamp(context) return super().get_url_params(context, next_page_token) - def get_next_page_token( - self, - response: requests.Response, - previous_token: Any | None, # noqa: ANN401 - ) -> Any | None: # noqa: ANN401 - """ - Exit early if first (oldest) record in the page is older than the replication - bookmark. With github's default record ordering, each page contains records - in ascending order. - """ - self.logger.debug("Cutoff: %s", self.cutoff) - if self.cutoff: - replies_jsonpath = ( - "$.data.repository.discussion.comments.nodes.[*].replies.nodes.[*]" - ) - results = list(extract_jsonpath(replies_jsonpath, input=response.json())) - if results: - oldest_created_at = parse(results[0][self.replication_key]) - if oldest_created_at < self.cutoff: - self.logger.info( - "Early exit: oldest=%s, cutoff=%s", - oldest_created_at, - self.cutoff, + def get_new_paginator(self): # noqa: ANN201 + """Get a new paginator for this stream.""" + stream = self + + class DiscussionRepliesPaginator(GitHubGraphQLPaginator): + def has_more(self, response) -> bool: # noqa: ANN001 + stream.logger.debug("Cutoff: %s", stream.cutoff) + if stream.cutoff: + replies_jsonpath = "$.data.repository.discussion.comments.nodes.[*].replies.nodes.[*]" # noqa: E501 + results = list( + extract_jsonpath(replies_jsonpath, input=response.json()) ) - return None # early exit - return super().get_next_page_token(response, previous_token) + if results: + oldest_created_at = parse(results[0][stream.replication_key]) + if oldest_created_at < stream.cutoff: + stream.logger.info( + "Early exit: oldest=%s, cutoff=%s", + oldest_created_at, + stream.cutoff, + ) + return False + return super().has_more(response) + + return DiscussionRepliesPaginator(stream) def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """Return a generator of row-type dictionary objects. From b35e942b7e772802bcff7aaf65675d4576ad7489 Mon Sep 17 00:00:00 2001 From: Atif Imam Date: Sun, 30 Aug 2026 19:17:03 +0530 Subject: [PATCH 2/6] fix: update stale comment referencing removed get_next_page_token method --- tap_github/client.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tap_github/client.py b/tap_github/client.py index 9efbfb87..948853c3 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -248,7 +248,7 @@ def get_url_params( params["direction"] = "desc" if self.use_fake_since_parameter else "asc" # Unfortunately the /starred, /stargazers (starred_at) and /events (created_at) endpoints do not support # noqa: E501 - # the "since" parameter out of the box. But we use a workaround in 'get_next_page_token'. # noqa: E501 + # the "since" parameter out of the box. But we use a workaround in the paginator. # noqa: E501 elif self.replication_key in ["starred_at", "created_at"]: params["sort"] = "created" params["direction"] = "desc" From 4d63dd0f6604441ba716a34402c64add9b7b9cd6 Mon Sep 17 00:00:00 2001 From: Atif Imam Date: Wed, 2 Sep 2026 21:48:20 +0530 Subject: [PATCH 3/6] fix: decouple paginators from stream objects Address review feedback by passing explicit parameters to paginator constructors instead of the stream object itself. - GitHubRestPaginator now takes keyword-only parameters derived from stream attributes (max_results_limit, max_per_page, use_cursor_pagination, replication_key, use_fake_since_parameter, records_jsonpath) and stores them as private state. - GitHubGraphQLPaginator no longer requires any constructor arguments. - RepositoryStream sets the search MAX_RESULTS_LIMIT in __init__ instead of as a side effect of the path property. - The four specialized GraphQL paginators capture only the values they need (logger, query_jsonpath, replication_key) and derive the incremental cutoff from the request URL, matching the existing StargazersPaginator behavior. - Add deterministic unit tests covering page-number and cursor pagination, result limits, fake-since and commit-timestamp early exits, nested GraphQL cursor handling, the specialized cutoff paths, and the absence of the legacy get_next_page_token deprecation warning. --- tap_github/client.py | 57 ++++-- tap_github/repository_streams.py | 136 ++++++------- tests/test_paginators.py | 329 +++++++++++++++++++++++++++++++ 3 files changed, 428 insertions(+), 94 deletions(-) create mode 100644 tests/test_paginators.py diff --git a/tap_github/client.py b/tap_github/client.py index 948853c3..24e4bfa6 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -32,19 +32,33 @@ class GitHubRestPaginator(BaseAPIPaginator[int | str | None]): """Paginator for GitHub REST API streams.""" - def __init__(self, stream: GitHubRestStream, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 - super().__init__(None, *args, **kwargs) - self.stream = stream + def __init__( + self, + *, + max_results_limit: int | None, + max_per_page: int, + use_cursor_pagination: bool, + replication_key: str | None, + use_fake_since_parameter: bool, + records_jsonpath: str, + ) -> None: + super().__init__(None) + self._max_results_limit = max_results_limit + self._max_per_page = max_per_page + self._use_cursor_pagination = use_cursor_pagination + self._replication_key = replication_key + self._use_fake_since_parameter = use_fake_since_parameter + self._records_jsonpath = records_jsonpath def has_more(self, response: requests.Response) -> bool: """Check if there are more pages.""" if ( self.current_value - and self.stream.MAX_RESULTS_LIMIT - and not self.stream.use_cursor_pagination + and self._max_results_limit + and not self._use_cursor_pagination and ( - cast("int", self.current_value) * self.stream.MAX_PER_PAGE - >= self.stream.MAX_RESULTS_LIMIT + cast("int", self.current_value) * self._max_per_page + >= self._max_results_limit ) ): return False @@ -56,13 +70,13 @@ def has_more(self, response: requests.Response) -> bool: results = ( resp_json if isinstance(resp_json, list) - else list(extract_jsonpath(self.stream.records_jsonpath, input=resp_json)) + else list(extract_jsonpath(self._records_jsonpath, input=resp_json)) ) if not results: return False - if self.stream.replication_key and self.stream.use_fake_since_parameter: + if self._replication_key and self._use_fake_since_parameter: request_parameters = parse_qs(str(urlparse(response.request.url).query)) try: since = ( @@ -80,8 +94,8 @@ def has_more(self, response: requests.Response) -> bool: ) replication_date = ( - results[-1][self.stream.replication_key] - if self.stream.replication_key != "commit_timestamp" + results[-1][self._replication_key] + if self._replication_key != "commit_timestamp" else results[-1]["commit"]["committer"]["date"] ) if ( @@ -95,7 +109,7 @@ def has_more(self, response: requests.Response) -> bool: def get_next(self, response: requests.Response) -> int | str | None: """Get the next pagination token.""" - if self.stream.use_cursor_pagination: + if self._use_cursor_pagination: parsed_url = urlparse(response.links["next"]["url"]) captured_after_value_list = parse_qs(parsed_url.query).get("after") return captured_after_value_list[0] if captured_after_value_list else None @@ -117,9 +131,8 @@ def get_next(self, response: requests.Response) -> int | str | None: class GitHubGraphQLPaginator(BaseAPIPaginator[dict[str, str] | None]): """Paginator for GitHub GraphQL API streams.""" - def __init__(self, stream: GitHubGraphqlStream, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 - super().__init__(None, *args, **kwargs) - self.stream = stream + def __init__(self) -> None: + super().__init__(None) def has_more(self, response: requests.Response) -> bool: """Check if there are more pages.""" @@ -228,7 +241,17 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]: def get_new_paginator(self) -> BaseAPIPaginator | None: """Get a new paginator for this stream.""" - return GitHubRestPaginator(self) + return GitHubRestPaginator( + max_results_limit=self.MAX_RESULTS_LIMIT, + max_per_page=self.MAX_PER_PAGE, + use_cursor_pagination=self.use_cursor_pagination, + replication_key=cast( # type: ignore[redundant-cast] + "str | None", + self.replication_key, + ), + use_fake_since_parameter=self.use_fake_since_parameter, + records_jsonpath=self.records_jsonpath, + ) def get_url_params( self, @@ -505,7 +528,7 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]: def get_new_paginator(self) -> BaseAPIPaginator | None: """Get a new paginator for this stream.""" - return GitHubGraphQLPaginator(self) + return GitHubGraphQLPaginator() def get_url_params( self, diff --git a/tap_github/repository_streams.py b/tap_github/repository_streams.py index d8fb5a6f..2f83acb7 100644 --- a/tap_github/repository_streams.py +++ b/tap_github/repository_streams.py @@ -30,7 +30,6 @@ if TYPE_CHECKING: from collections.abc import Iterable - from datetime import datetime import requests from singer_sdk import Tap @@ -47,6 +46,11 @@ class RepositoryStream(GitHubRestStream): # e.g. when the description or the primary language of the repository is updated. replication_key = "updated_at" + def __init__(self, *args: Any, **kwargs: Any) -> None: # noqa: ANN401 + super().__init__(*args, **kwargs) + if "searches" in self.config: + self.MAX_RESULTS_LIMIT = 1000 + def get_url_params( self, context: Context | None, @@ -66,8 +70,6 @@ def path(self) -> str: # type: ignore[override, return] # ty:ignore[invalid-ret """Return the API endpoint path. Path options are mutually exclusive.""" if "searches" in self.config: - # Search API max: 1,000 total. - self.MAX_RESULTS_LIMIT = 1000 return "/search/repositories" if "repositories" in self.config: # the `repo` and `org` args will be parsed from the partition's `context` @@ -2069,12 +2071,12 @@ def post_process(self, row: dict, context: Context | None = None) -> dict: row["user_id"] = row["user"]["id"] return row - def get_new_paginator(self): # noqa: ANN201 + def get_new_paginator(self) -> GitHubGraphQLPaginator: """Get a new paginator for this stream.""" - stream = self + query_jsonpath = self.query_jsonpath class StargazersPaginator(GitHubGraphQLPaginator): - def has_more(self, response) -> bool: # noqa: ANN001 + def has_more(self, response: requests.Response) -> bool: request_parameters = parse_qs(str(urlparse(response.request.url).query)) try: since = ( @@ -2086,7 +2088,7 @@ def has_more(self, response) -> bool: # noqa: ANN001 since = "" if since: results = list( - extract_jsonpath(stream.query_jsonpath, input=response.json()) + extract_jsonpath(query_jsonpath, input=response.json()) ) if len(results) == 0: return False @@ -2095,7 +2097,7 @@ def has_more(self, response) -> bool: # noqa: ANN001 return False return super().has_more(response) - return StargazersPaginator(stream) + return StargazersPaginator() @property def query(self) -> str: @@ -2266,41 +2268,34 @@ class DiscussionsStream(GitHubGraphqlStream): ignore_parent_replication_key = True # Repository's updated_at does not change when a new discussion is added # noqa: E501 is_sorted = False # Singer recognizes as unsorted. - def __init__(self, *args, **kwargs) -> None: # noqa: ANN002, ANN003 - super().__init__(*args, **kwargs) - self.cutoff: datetime | None = None - - def get_url_params( - self, - context: Context | None, - next_page_token: Any | None, # noqa: ANN401 - ) -> dict[str, Any]: - self.cutoff = self.get_starting_timestamp(context) - return super().get_url_params(context, next_page_token) - - def get_new_paginator(self): # noqa: ANN201 + def get_new_paginator(self) -> GitHubGraphQLPaginator: """Get a new paginator for this stream.""" - stream = self + logger = self.logger + query_jsonpath = self.query_jsonpath + replication_key = self.replication_key class DiscussionsPaginator(GitHubGraphQLPaginator): - def has_more(self, response) -> bool: # noqa: ANN001 - stream.logger.debug("Cutoff: %s", stream.cutoff) - if stream.cutoff: + def has_more(self, response: requests.Response) -> bool: + request_parameters = parse_qs(str(urlparse(response.request.url).query)) + cutoff_str = request_parameters.get("since", [None])[0] + cutoff = parse(cutoff_str) if cutoff_str else None + logger.debug("Cutoff: %s", cutoff) + if cutoff: results = list( - extract_jsonpath(stream.query_jsonpath, input=response.json()) + extract_jsonpath(query_jsonpath, input=response.json()) ) if results: - oldest_updated_at = parse(results[-1][stream.replication_key]) - if oldest_updated_at < stream.cutoff: - stream.logger.info( + oldest_updated_at = parse(results[-1][replication_key]) + if oldest_updated_at < cutoff: + logger.info( "Early exit: oldest=%s, cutoff=%s", oldest_updated_at, - stream.cutoff, + cutoff, ) return False return super().has_more(response) - return DiscussionsPaginator(stream) + return DiscussionsPaginator() def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """ @@ -2588,41 +2583,34 @@ class DiscussionCommentsStream(GitHubGraphqlStream): is_sorted = False # Set as False to avoid data loss. # If treated as sorted, Singer will bookmark state as page-1's first record and skip older pages on incremental runs (data loss). # noqa: E501 - def __init__(self, *args, **kwargs) -> None: # noqa: ANN002, ANN003 - super().__init__(*args, **kwargs) - self.cutoff: datetime | None = None - - def get_url_params( - self, - context: Context | None, - next_page_token: Any | None, # noqa: ANN401 - ) -> dict[str, Any]: - self.cutoff = self.get_starting_timestamp(context) - return super().get_url_params(context, next_page_token) - - def get_new_paginator(self): # noqa: ANN201 + def get_new_paginator(self) -> GitHubGraphQLPaginator: """Get a new paginator for this stream.""" - stream = self + logger = self.logger + query_jsonpath = self.query_jsonpath + replication_key = self.replication_key class DiscussionCommentsPaginator(GitHubGraphQLPaginator): - def has_more(self, response) -> bool: # noqa: ANN001 - stream.logger.debug("Cutoff: %s", stream.cutoff) - if stream.cutoff: + def has_more(self, response: requests.Response) -> bool: + request_parameters = parse_qs(str(urlparse(response.request.url).query)) + cutoff_str = request_parameters.get("since", [None])[0] + cutoff = parse(cutoff_str) if cutoff_str else None + logger.debug("Cutoff: %s", cutoff) + if cutoff: results = list( - extract_jsonpath(stream.query_jsonpath, input=response.json()) + extract_jsonpath(query_jsonpath, input=response.json()) ) if results: - oldest_created_at = parse(results[0][stream.replication_key]) - if oldest_created_at < stream.cutoff: - stream.logger.info( + oldest_created_at = parse(results[0][replication_key]) + if oldest_created_at < cutoff: + logger.info( "Early exit: oldest=%s, cutoff=%s", oldest_created_at, - stream.cutoff, + cutoff, ) return False return super().has_more(response) - return DiscussionCommentsPaginator(stream) + return DiscussionCommentsPaginator() def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """ @@ -2849,42 +2837,36 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]: reply["comment_id"] = comment_id yield reply - def __init__(self, *args, **kwargs) -> None: # noqa: ANN002, ANN003 - super().__init__(*args, **kwargs) - self.cutoff: datetime | None = None - - def get_url_params( - self, - context: Context | None, - next_page_token: Any | None, # noqa: ANN401 - ) -> dict[str, Any]: - self.cutoff = self.get_starting_timestamp(context) - return super().get_url_params(context, next_page_token) - - def get_new_paginator(self): # noqa: ANN201 + def get_new_paginator(self) -> GitHubGraphQLPaginator: """Get a new paginator for this stream.""" - stream = self + logger = self.logger + replication_key = self.replication_key + replies_jsonpath = ( + "$.data.repository.discussion.comments.nodes.[*].replies.nodes.[*]" + ) class DiscussionRepliesPaginator(GitHubGraphQLPaginator): - def has_more(self, response) -> bool: # noqa: ANN001 - stream.logger.debug("Cutoff: %s", stream.cutoff) - if stream.cutoff: - replies_jsonpath = "$.data.repository.discussion.comments.nodes.[*].replies.nodes.[*]" # noqa: E501 + def has_more(self, response: requests.Response) -> bool: + request_parameters = parse_qs(str(urlparse(response.request.url).query)) + cutoff_str = request_parameters.get("since", [None])[0] + cutoff = parse(cutoff_str) if cutoff_str else None + logger.debug("Cutoff: %s", cutoff) + if cutoff: results = list( extract_jsonpath(replies_jsonpath, input=response.json()) ) if results: - oldest_created_at = parse(results[0][stream.replication_key]) - if oldest_created_at < stream.cutoff: - stream.logger.info( + oldest_created_at = parse(results[0][replication_key]) + if oldest_created_at < cutoff: + logger.info( "Early exit: oldest=%s, cutoff=%s", oldest_created_at, - stream.cutoff, + cutoff, ) return False return super().has_more(response) - return DiscussionRepliesPaginator(stream) + return DiscussionRepliesPaginator() def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """Return a generator of row-type dictionary objects. diff --git a/tests/test_paginators.py b/tests/test_paginators.py new file mode 100644 index 00000000..3d2345ee --- /dev/null +++ b/tests/test_paginators.py @@ -0,0 +1,329 @@ +"""Tests for the custom REST and GraphQL paginators.""" + +from __future__ import annotations + +import warnings +from typing import ClassVar +from unittest.mock import MagicMock + +import pytest + +from tap_github.client import ( + GitHubGraphQLPaginator, + GitHubRestPaginator, + GitHubRestStream, +) +from tap_github.repository_streams import ( + DiscussionCommentRepliesStream, + DiscussionCommentsStream, + DiscussionsStream, + RepositoryStream, + StargazersGraphqlStream, +) +from tap_github.tap import TapGitHub + + +def _response( + json_data=None, + *, + next_url: str | None = None, + request_url: str = "https://api.github.com/test?page=1&per_page=100", +) -> MagicMock: + response = MagicMock() + response.json.return_value = json_data if json_data is not None else [] + response.links = {"next": {"url": next_url}} if next_url else {} + response.request.url = request_url + return response + + +def _rest_paginator(**overrides) -> GitHubRestPaginator: + kwargs = { + "max_results_limit": None, + "max_per_page": 100, + "use_cursor_pagination": False, + "replication_key": None, + "use_fake_since_parameter": False, + "records_jsonpath": "$[*]", + } + kwargs.update(overrides) + return GitHubRestPaginator(**kwargs) + + +def _tap(config: dict | None = None) -> TapGitHub: + return TapGitHub(config=config or {"repositories": ["org/repo"]}) + + +def _nested(keys: tuple[str, ...], value: dict) -> dict: + result = value + for key in reversed(keys): + result = {key: result} + return result + + +def _graphql_page( + collection_path: tuple[str, ...], + records_key: str, + records: list[dict], +) -> dict: + collection = { + "pageInfo": {"hasNextPage_0": True, "endCursor_0": "next"}, + records_key: records, + } + return _nested(("data", "repository", *collection_path), collection) + + +class _TestRestStream(GitHubRestStream): + name = "test_rest" + path = "/test" + schema: ClassVar[dict] = { + "type": "object", + "properties": {"id": {"type": "integer"}}, + } + replication_key = "updated_at" + use_fake_since_parameter = True + + +@pytest.mark.parametrize( + ("current", "next_url", "expected"), + [ + (None, "https://api.github.com/test?page=2", 2), + (3, "https://api.github.com/test?per_page=100", 4), + (3, "https://api.github.com/test?page=8", 8), + ], +) +def test_rest_page_number_tokens(current, next_url, expected): + paginator = _rest_paginator() + paginator._value = current + + assert paginator.get_next(_response(next_url=next_url)) == expected + + +@pytest.mark.parametrize( + ("next_url", "expected"), + [ + ("https://api.github.com/test?after=cursor-2", "cursor-2"), + ("https://api.github.com/test?page=2", None), + ], +) +def test_rest_cursor_tokens(next_url, expected): + paginator = _rest_paginator(use_cursor_pagination=True) + + assert paginator.get_next(_response(next_url=next_url)) == expected + + +@pytest.mark.parametrize( + ("json_data", "next_url", "expected"), + [ + ([{"id": 1}], "next", True), + ([], "next", False), + ([{"id": 1}], None, False), + ], +) +def test_rest_has_more_requires_records_and_a_next_link( + json_data, + next_url, + expected, +): + paginator = _rest_paginator() + + assert paginator.has_more(_response(json_data, next_url=next_url)) is expected + + +@pytest.mark.parametrize( + ("current", "limit", "cursor_mode", "expected"), + [ + (2, 200, False, False), + (2, 300, False, True), + ("cursor", 100, True, True), + ], +) +def test_rest_max_results_limit(current, limit, cursor_mode, expected): + paginator = _rest_paginator( + max_results_limit=limit, + use_cursor_pagination=cursor_mode, + ) + paginator._value = current + response = _response([{"id": 1}], next_url="next") + + assert paginator.has_more(response) is expected + + +@pytest.mark.parametrize( + ("record_date", "since", "expected"), + [ + ("2024-06-01T00:00:00Z", "2025-01-01T00:00:00Z", False), + ("2025-06-01T00:00:00Z", "2025-01-01T00:00:00Z", True), + ("2025-06-01T00:00:00Z", None, True), + ], +) +def test_rest_fake_since_early_exit(record_date, since, expected): + paginator = _rest_paginator( + replication_key="starred_at", + use_fake_since_parameter=True, + ) + query = "direction=desc" + if since: + query = f"fake_since={since}&{query}" + response = _response( + [{"starred_at": record_date}], + next_url="next", + request_url=f"https://api.github.com/test?{query}", + ) + + assert paginator.has_more(response) is expected + + +def test_rest_fake_since_supports_commit_timestamp(): + paginator = _rest_paginator( + replication_key="commit_timestamp", + use_fake_since_parameter=True, + ) + response = _response( + [{"commit": {"committer": {"date": "2024-01-01T00:00:00Z"}}}], + next_url="next", + request_url=( + "https://api.github.com/test?fake_since=2025-01-01T00:00:00Z&direction=desc" + ), + ) + + assert paginator.has_more(response) is False + + +def test_rest_has_more_uses_records_jsonpath(): + paginator = _rest_paginator(records_jsonpath="$.items[*]") + + assert paginator.has_more(_response({"items": [{"id": 1}]}, next_url="next")) + + +def test_graphql_paginator_advances_deepest_cursor(): + paginator = GitHubGraphQLPaginator() + paginator._value = { + "nextPageCursor_0": "outer", + "nextPageCursor_1": "old-inner", + } + response = _response( + { + "data": { + "pageInfo": { + "hasNextPage_0": False, + "endCursor_0": "unused", + }, + "nested": { + "pageInfo": { + "hasNextPage_1": True, + "endCursor_1": "new-inner", + } + }, + } + } + ) + + assert paginator.has_more(response) is True + assert paginator.get_next(response) == { + "nextPageCursor_0": "outer", + "nextPageCursor_1": "new-inner", + } + + +def test_graphql_paginator_finishes_without_next_page(): + paginator = GitHubGraphQLPaginator() + response = _response({"pageInfo": {"hasNextPage_0": False}}) + + assert paginator.has_more(response) is False + assert paginator.get_next(response) is None + + +def test_rest_stream_passes_explicit_attributes_to_paginator(): + paginator = _TestRestStream(_tap()).get_new_paginator() + + assert isinstance(paginator, GitHubRestPaginator) + assert paginator._max_results_limit is None + assert paginator._max_per_page == 100 + assert paginator._use_cursor_pagination is False + assert paginator._replication_key == "updated_at" + assert paginator._use_fake_since_parameter is True + assert paginator._records_jsonpath == "$[*]" + assert not hasattr(paginator, "stream") + + +def test_repository_search_limit_is_available_before_path_access(): + stream = RepositoryStream( + _tap({"searches": [{"name": "taps", "query": "topic:singer-tap"}]}) + ) + + paginator = stream.get_new_paginator() + + assert isinstance(paginator, GitHubRestPaginator) + assert paginator._max_results_limit == 1000 + assert not hasattr(paginator, "stream") + + +def test_request_records_uses_new_paginator_without_legacy_warning(): + stream = _TestRestStream(_tap()) + first_response = _response( + [{"id": 1, "updated_at": "2025-01-01T00:00:00Z"}], + next_url="https://api.github.com/test?page=2&per_page=100", + ) + second_response = _response([{"id": 2, "updated_at": "2025-01-02T00:00:00Z"}]) + stream._request = MagicMock(side_effect=[first_response, second_response]) + + with warnings.catch_warnings(record=True) as caught: + warnings.simplefilter("always") + records = list(stream.request_records(None)) + + assert [record["id"] for record in records] == [1, 2] + assert stream._request.call_count == 2 + second_request = stream._request.call_args_list[1].args[0] + assert "page=2" in second_request.url + assert not any("get_next_page_token" in str(item.message) for item in caught) + + +@pytest.mark.parametrize( + ("stream_type", "response_json"), + [ + ( + StargazersGraphqlStream, + _graphql_page( + ("stargazers",), + "edges", + [{"starred_at": "2024-01-01T00:00:00Z"}], + ), + ), + ( + DiscussionsStream, + _graphql_page( + ("discussions",), + "nodes", + [{"updated_at": "2024-01-01T00:00:00Z"}], + ), + ), + ( + DiscussionCommentsStream, + _graphql_page( + ("discussion", "comments"), + "nodes", + [{"created_at": "2024-01-01T00:00:00Z"}], + ), + ), + ( + DiscussionCommentRepliesStream, + _graphql_page( + ("discussion", "comments"), + "nodes", + [{"replies": {"nodes": [{"created_at": "2024-01-01T00:00:00Z"}]}}], + ), + ), + ], +) +def test_graphql_stream_paginators_preserve_incremental_early_exit( + stream_type, + response_json, +): + paginator = stream_type(_tap()).get_new_paginator() + response = _response( + response_json, + request_url="https://api.github.com/graphql?since=2025-01-01T00:00:00Z", + ) + + assert paginator.has_more(response) is False + assert not hasattr(paginator, "stream") From 2bb32d1e3974afe0dd48b543fe172b23b91ecd9c Mon Sep 17 00:00:00 2001 From: Atif Imam Date: Thu, 3 Sep 2026 12:19:36 +0530 Subject: [PATCH 4/6] fix: remove duplicate GraphQL pagination check --- tap_github/client.py | 16 ---------------- tests/test_paginators.py | 11 +++++++---- 2 files changed, 7 insertions(+), 20 deletions(-) diff --git a/tap_github/client.py b/tap_github/client.py index fb4213bf..c0fd222a 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -132,22 +132,6 @@ class GitHubGraphQLPaginator(BaseAPIPaginator[dict[str, str] | None]): def __init__(self) -> None: super().__init__(None) - def has_more(self, response: requests.Response) -> bool: - """Check if there are more pages.""" - resp_json = response.json() - next_page_results = nested_lookup( - key="hasNextPage_", - document=resp_json, - wild=True, - with_keys=True, - ) - has_next_page_indices: list[int] = [] - for key, value in next_page_results.items(): - if any(value): - pagination_index = int(str(key).split("_")[1]) - has_next_page_indices.append(pagination_index) - return len(has_next_page_indices) > 0 - def get_next(self, response: requests.Response) -> dict[str, str] | None: """Get the next pagination token.""" resp_json = response.json() diff --git a/tests/test_paginators.py b/tests/test_paginators.py index 3d2345ee..00e4b790 100644 --- a/tests/test_paginators.py +++ b/tests/test_paginators.py @@ -218,19 +218,22 @@ def test_graphql_paginator_advances_deepest_cursor(): } ) - assert paginator.has_more(response) is True - assert paginator.get_next(response) == { + paginator.advance(response) + + assert paginator.current_value == { "nextPageCursor_0": "outer", "nextPageCursor_1": "new-inner", } + assert paginator.finished is False def test_graphql_paginator_finishes_without_next_page(): paginator = GitHubGraphQLPaginator() response = _response({"pageInfo": {"hasNextPage_0": False}}) - assert paginator.has_more(response) is False - assert paginator.get_next(response) is None + paginator.advance(response) + + assert paginator.finished is True def test_rest_stream_passes_explicit_attributes_to_paginator(): From a9034c59980b2779ed6c427bc151cb56dc4444da Mon Sep 17 00:00:00 2001 From: Atif Imam Date: Thu, 3 Sep 2026 12:26:06 +0530 Subject: [PATCH 5/6] refactor: parameterize GraphQL cutoff pagination --- tap_github/client.py | 89 +++++++++++++++++---- tap_github/repository_streams.py | 129 +++---------------------------- tests/test_paginators.py | 9 ++- 3 files changed, 88 insertions(+), 139 deletions(-) diff --git a/tap_github/client.py b/tap_github/client.py index c0fd222a..e2ea9b87 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -19,6 +19,7 @@ if TYPE_CHECKING: from collections.abc import Iterable + from logging import Logger import requests from backoff.types import Details @@ -27,6 +28,20 @@ EMPTY_REPO_ERROR_STATUS = 409 +def _get_request_query_parameter( + response: requests.Response, + parameter: str, +) -> str | None: + """Return a query parameter from the URL used for a response.""" + values = parse_qs(urlparse(response.request.url or "").query).get(parameter) + if not values: + return None + + # parse_qs interprets a literal "+" as a space. Restore it so timestamps + # remain timezone-aware whether the URL contains "+" or "%2B". + return values[0].replace(" ", "+") + + class GitHubRestPaginator(BaseAPIPaginator[int | str | None]): """Paginator for GitHub REST API streams.""" @@ -75,21 +90,8 @@ def has_more(self, response: requests.Response) -> bool: return False if self._replication_key and self._use_fake_since_parameter: - request_parameters = parse_qs(str(urlparse(response.request.url).query)) - try: - since = ( - request_parameters["fake_since"][0].replace(" ", "+") - if "fake_since" in request_parameters - else "" - ) - except IndexError: - return False - - direction = ( - request_parameters["direction"][0] - if "direction" in request_parameters - else None - ) + since = _get_request_query_parameter(response, "fake_since") + direction = _get_request_query_parameter(response, "direction") replication_date = ( results[-1][self._replication_key] @@ -129,12 +131,53 @@ def get_next(self, response: requests.Response) -> int | str | None: class GitHubGraphQLPaginator(BaseAPIPaginator[dict[str, str] | None]): """Paginator for GitHub GraphQL API streams.""" - def __init__(self) -> None: + def __init__( + self, + *, + cutoff_jsonpath: str | None = None, + replication_key: str | None = None, + records_are_ascending: bool = False, + logger: Logger | None = None, + ) -> None: super().__init__(None) + self._cutoff_jsonpath = cutoff_jsonpath + self._replication_key = replication_key + self._records_are_ascending = records_are_ascending + self._logger = logger + + def _reached_cutoff( + self, + response: requests.Response, + response_json: dict[str, Any], + ) -> bool: + """Return whether the current page has reached the replication cutoff.""" + since = _get_request_query_parameter(response, "since") + if not since or not self._cutoff_jsonpath or not self._replication_key: + return False + + cutoff = parse(since) + if self._logger: + self._logger.debug("Cutoff: %s", cutoff) + + records = list(extract_jsonpath(self._cutoff_jsonpath, input=response_json)) + if not records: + return False + + oldest_record = records[0] if self._records_are_ascending else records[-1] + oldest = parse(oldest_record[self._replication_key]) + if oldest >= cutoff: + return False + + if self._logger: + self._logger.info("Early exit: oldest=%s, cutoff=%s", oldest, cutoff) + return True def get_next(self, response: requests.Response) -> dict[str, str] | None: """Get the next pagination token.""" resp_json = response.json() + if self._reached_cutoff(response, resp_json): + return None + next_page_results = nested_lookup( key="hasNextPage_", document=resp_json, @@ -482,6 +525,9 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]: class GitHubGraphqlStream(GraphQLStream, GitHubRestStream): """GitHub Graphql stream class.""" + pagination_cutoff_jsonpath: str | None = None + pagination_records_are_ascending = False + @property def url_base(self) -> str: return f"{self.config.get('api_url_base', self.DEFAULT_API_BASE_URL)}/graphql" @@ -508,7 +554,16 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]: def get_new_paginator(self) -> BaseAPIPaginator | None: """Get a new paginator for this stream.""" - return GitHubGraphQLPaginator() + replication_key = self.replication_key + if not isinstance(replication_key, str): + replication_key = None + + return GitHubGraphQLPaginator( + cutoff_jsonpath=self.pagination_cutoff_jsonpath, + replication_key=replication_key, + records_are_ascending=self.pagination_records_are_ascending, + logger=self.logger, + ) def get_url_params( self, diff --git a/tap_github/repository_streams.py b/tap_github/repository_streams.py index 2f83acb7..036cc1cc 100644 --- a/tap_github/repository_streams.py +++ b/tap_github/repository_streams.py @@ -5,16 +5,14 @@ import http from collections import defaultdict from typing import TYPE_CHECKING, Any, ClassVar -from urllib.parse import parse_qs, urlparse +from urllib.parse import urlparse -from dateutil.parser import parse from singer_sdk import typing as th # JSON Schema typing helpers from singer_sdk.exceptions import FatalAPIError, RetriableAPIError from singer_sdk.helpers.jsonpath import extract_jsonpath from tap_github.client import ( GitHubDiffStream, - GitHubGraphQLPaginator, GitHubGraphqlStream, GitHubRestStream, ) @@ -2050,6 +2048,7 @@ class StargazersGraphqlStream(GitHubGraphqlStream): query_jsonpath = "$.data.repository.stargazers.edges.[*]" primary_keys: ClassVar[list[str]] = ["user_id", "repo_id"] replication_key = "starred_at" + pagination_cutoff_jsonpath = query_jsonpath parent_stream_type = RepositoryStream state_partitioning_keys: ClassVar[list[str]] = ["repo_id"] # The parent repository object changes if the number of stargazers changes. @@ -2071,34 +2070,6 @@ def post_process(self, row: dict, context: Context | None = None) -> dict: row["user_id"] = row["user"]["id"] return row - def get_new_paginator(self) -> GitHubGraphQLPaginator: - """Get a new paginator for this stream.""" - query_jsonpath = self.query_jsonpath - - class StargazersPaginator(GitHubGraphQLPaginator): - def has_more(self, response: requests.Response) -> bool: - request_parameters = parse_qs(str(urlparse(response.request.url).query)) - try: - since = ( - request_parameters["since"][0].replace(" ", "+") - if "since" in request_parameters - else "" - ) - except IndexError: - since = "" - if since: - results = list( - extract_jsonpath(query_jsonpath, input=response.json()) - ) - if len(results) == 0: - return False - last = results[-1] - if parse(last["starred_at"]) < parse(since): - return False - return super().has_more(response) - - return StargazersPaginator() - @property def query(self) -> str: """Return dynamic GraphQL query.""" @@ -2263,40 +2234,12 @@ class DiscussionsStream(GitHubGraphqlStream): "id" ] # databaseId renamed to id to keep tap consistent with REST streams. replication_key = "updated_at" + pagination_cutoff_jsonpath = query_jsonpath parent_stream_type = RepositoryStream state_partitioning_keys: ClassVar[list[str]] = ["repo_id"] ignore_parent_replication_key = True # Repository's updated_at does not change when a new discussion is added # noqa: E501 is_sorted = False # Singer recognizes as unsorted. - def get_new_paginator(self) -> GitHubGraphQLPaginator: - """Get a new paginator for this stream.""" - logger = self.logger - query_jsonpath = self.query_jsonpath - replication_key = self.replication_key - - class DiscussionsPaginator(GitHubGraphQLPaginator): - def has_more(self, response: requests.Response) -> bool: - request_parameters = parse_qs(str(urlparse(response.request.url).query)) - cutoff_str = request_parameters.get("since", [None])[0] - cutoff = parse(cutoff_str) if cutoff_str else None - logger.debug("Cutoff: %s", cutoff) - if cutoff: - results = list( - extract_jsonpath(query_jsonpath, input=response.json()) - ) - if results: - oldest_updated_at = parse(results[-1][replication_key]) - if oldest_updated_at < cutoff: - logger.info( - "Early exit: oldest=%s, cutoff=%s", - oldest_updated_at, - cutoff, - ) - return False - return super().has_more(response) - - return DiscussionsPaginator() - def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """ Return a generator of row-type dictionary objects. @@ -2578,40 +2521,13 @@ class DiscussionCommentsStream(GitHubGraphqlStream): "id" ] # databaseId renamed to id to keep tap consistent with REST streams. replication_key = "created_at" # API's default record ordering field. + pagination_cutoff_jsonpath = query_jsonpath + pagination_records_are_ascending = True parent_stream_type = DiscussionsStream state_partitioning_keys: ClassVar[list[str]] = ["discussion_id"] is_sorted = False # Set as False to avoid data loss. # If treated as sorted, Singer will bookmark state as page-1's first record and skip older pages on incremental runs (data loss). # noqa: E501 - def get_new_paginator(self) -> GitHubGraphQLPaginator: - """Get a new paginator for this stream.""" - logger = self.logger - query_jsonpath = self.query_jsonpath - replication_key = self.replication_key - - class DiscussionCommentsPaginator(GitHubGraphQLPaginator): - def has_more(self, response: requests.Response) -> bool: - request_parameters = parse_qs(str(urlparse(response.request.url).query)) - cutoff_str = request_parameters.get("since", [None])[0] - cutoff = parse(cutoff_str) if cutoff_str else None - logger.debug("Cutoff: %s", cutoff) - if cutoff: - results = list( - extract_jsonpath(query_jsonpath, input=response.json()) - ) - if results: - oldest_created_at = parse(results[0][replication_key]) - if oldest_created_at < cutoff: - logger.info( - "Early exit: oldest=%s, cutoff=%s", - oldest_created_at, - cutoff, - ) - return False - return super().has_more(response) - - return DiscussionCommentsPaginator() - def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """ Return a generator of row-type dictionary objects. @@ -2818,6 +2734,10 @@ class DiscussionCommentRepliesStream(GitHubGraphqlStream): "id" ] # databaseId renamed to id to keep tap consistent with REST streams. replication_key = "created_at" # API's default record ordering field. + pagination_cutoff_jsonpath = ( + "$.data.repository.discussion.comments.nodes.[*].replies.nodes.[*]" + ) + pagination_records_are_ascending = True parent_stream_type = DiscussionsStream # Only Discussion's timestamp is affected by replies. # noqa: E501 state_partitioning_keys: ClassVar[list[str]] = ["discussion_id"] is_sorted = False # Set as False to avoid data loss. @@ -2837,37 +2757,6 @@ def parse_response(self, response: requests.Response) -> Iterable[dict]: reply["comment_id"] = comment_id yield reply - def get_new_paginator(self) -> GitHubGraphQLPaginator: - """Get a new paginator for this stream.""" - logger = self.logger - replication_key = self.replication_key - replies_jsonpath = ( - "$.data.repository.discussion.comments.nodes.[*].replies.nodes.[*]" - ) - - class DiscussionRepliesPaginator(GitHubGraphQLPaginator): - def has_more(self, response: requests.Response) -> bool: - request_parameters = parse_qs(str(urlparse(response.request.url).query)) - cutoff_str = request_parameters.get("since", [None])[0] - cutoff = parse(cutoff_str) if cutoff_str else None - logger.debug("Cutoff: %s", cutoff) - if cutoff: - results = list( - extract_jsonpath(replies_jsonpath, input=response.json()) - ) - if results: - oldest_created_at = parse(results[0][replication_key]) - if oldest_created_at < cutoff: - logger.info( - "Early exit: oldest=%s, cutoff=%s", - oldest_created_at, - cutoff, - ) - return False - return super().has_more(response) - - return DiscussionRepliesPaginator() - def get_records(self, context: Context | None = None) -> Iterable[dict[str, Any]]: """Return a generator of row-type dictionary objects. If the parent discussion has no comments, skip the replies API call. diff --git a/tests/test_paginators.py b/tests/test_paginators.py index 00e4b790..fc67493c 100644 --- a/tests/test_paginators.py +++ b/tests/test_paginators.py @@ -325,8 +325,13 @@ def test_graphql_stream_paginators_preserve_incremental_early_exit( paginator = stream_type(_tap()).get_new_paginator() response = _response( response_json, - request_url="https://api.github.com/graphql?since=2025-01-01T00:00:00Z", + request_url=( + "https://api.github.com/graphql?since=2025-01-01T00%3A00%3A00%2B00%3A00" + ), ) - assert paginator.has_more(response) is False + paginator.advance(response) + + assert paginator.finished is True + assert type(paginator) is GitHubGraphQLPaginator assert not hasattr(paginator, "stream") From 5f478ca5b2271a98af8b554ed749051cb4e8cdd4 Mon Sep 17 00:00:00 2001 From: Atif Imam Date: Thu, 3 Sep 2026 18:01:10 +0530 Subject: [PATCH 6/6] fix: simplify paginator replication key handling --- tap_github/client.py | 9 +++++---- tests/test_paginators.py | 26 ++++++++++++++++++++++---- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/tap_github/client.py b/tap_github/client.py index e2ea9b87..1306179d 100644 --- a/tap_github/client.py +++ b/tap_github/client.py @@ -266,14 +266,15 @@ def get_records(self, context: Context | None) -> Iterable[dict[str, Any]]: def get_new_paginator(self) -> BaseAPIPaginator | None: """Get a new paginator for this stream.""" + replication_key = self.replication_key + if not isinstance(replication_key, str): + replication_key = None + return GitHubRestPaginator( max_results_limit=self.MAX_RESULTS_LIMIT, max_per_page=self.MAX_PER_PAGE, use_cursor_pagination=self.use_cursor_pagination, - replication_key=cast( # type: ignore[redundant-cast] - "str | None", - self.replication_key, - ), + replication_key=replication_key, use_fake_since_parameter=self.use_fake_since_parameter, records_jsonpath=self.records_jsonpath, ) diff --git a/tests/test_paginators.py b/tests/test_paginators.py index fc67493c..0f97aa2d 100644 --- a/tests/test_paginators.py +++ b/tests/test_paginators.py @@ -289,7 +289,10 @@ def test_request_records_uses_new_paginator_without_legacy_warning(): _graphql_page( ("stargazers",), "edges", - [{"starred_at": "2024-01-01T00:00:00Z"}], + [ + {"starred_at": "2026-01-01T00:00:00Z"}, + {"starred_at": "2024-01-01T00:00:00Z"}, + ], ), ), ( @@ -297,7 +300,10 @@ def test_request_records_uses_new_paginator_without_legacy_warning(): _graphql_page( ("discussions",), "nodes", - [{"updated_at": "2024-01-01T00:00:00Z"}], + [ + {"updated_at": "2026-01-01T00:00:00Z"}, + {"updated_at": "2024-01-01T00:00:00Z"}, + ], ), ), ( @@ -305,7 +311,10 @@ def test_request_records_uses_new_paginator_without_legacy_warning(): _graphql_page( ("discussion", "comments"), "nodes", - [{"created_at": "2024-01-01T00:00:00Z"}], + [ + {"created_at": "2024-01-01T00:00:00Z"}, + {"created_at": "2026-01-01T00:00:00Z"}, + ], ), ), ( @@ -313,7 +322,16 @@ def test_request_records_uses_new_paginator_without_legacy_warning(): _graphql_page( ("discussion", "comments"), "nodes", - [{"replies": {"nodes": [{"created_at": "2024-01-01T00:00:00Z"}]}}], + [ + { + "replies": { + "nodes": [ + {"created_at": "2024-01-01T00:00:00Z"}, + {"created_at": "2026-01-01T00:00:00Z"}, + ] + } + } + ], ), ), ],