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.

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

fix(templates): Use get_new_paginator in REST tap cookiecutter template #1646

Merged
Original file line number Diff line number Diff line change
Expand Up @@ -12,24 +12,29 @@ import requests
{% if cookiecutter.auth_method == "API Key" -%}
from singer_sdk.authenticators import APIKeyAuthenticator
from singer_sdk.helpers.jsonpath import extract_jsonpath
from singer_sdk.pagination import BaseAPIPaginator # noqa: TCH002
from singer_sdk.streams import {{ cookiecutter.stream_type }}Stream

{% elif cookiecutter.auth_method == "Bearer Token" -%}
from singer_sdk.authenticators import BearerTokenAuthenticator
from singer_sdk.helpers.jsonpath import extract_jsonpath
from singer_sdk.pagination import BaseAPIPaginator # noqa: TCH002
from singer_sdk.streams import {{ cookiecutter.stream_type }}Stream

{% elif cookiecutter.auth_method == "Basic Auth" -%}
from singer_sdk.authenticators import BasicAuthenticator
from singer_sdk.helpers.jsonpath import extract_jsonpath
from singer_sdk.pagination import BaseAPIPaginator # noqa: TCH002
from singer_sdk.streams import {{ cookiecutter.stream_type }}Stream

{% elif cookiecutter.auth_method == "Custom or N/A" -%}
from singer_sdk.helpers.jsonpath import extract_jsonpath
from singer_sdk.pagination import BaseAPIPaginator # noqa: TCH002
from singer_sdk.streams import {{ cookiecutter.stream_type }}Stream

{% elif cookiecutter.auth_method in ("OAuth2", "JWT") -%}
from singer_sdk.helpers.jsonpath import extract_jsonpath
from singer_sdk.pagination import BaseAPIPaginator # noqa: TCH002
from singer_sdk.streams import {{ cookiecutter.stream_type }}Stream

from {{ cookiecutter.library_name }}.auth import {{ cookiecutter.source_name }}Authenticator
Expand Down Expand Up @@ -59,7 +64,7 @@ class {{ cookiecutter.source_name }}Stream({{ cookiecutter.stream_type }}Stream)

records_jsonpath = "$[*]" # Or override `parse_response`.

# Set this value or override `get_next_page_token`.
# Set this value or override `get_new_paginator`.
next_page_token_jsonpath = "$.next_page" # noqa: S105

{%- if cookiecutter.auth_method in ("OAuth2", "JWT") %}
Expand Down Expand Up @@ -136,34 +141,20 @@ class {{ cookiecutter.source_name }}Stream({{ cookiecutter.stream_type }}Stream)
{%- endif %}
return headers

def get_next_page_token(
self,
response: requests.Response,
previous_token: Any | None, # noqa: ARG002
) -> Any | None:
"""Return a token for identifying next page or None if no more pages.
def get_new_paginator(self) -> BaseAPIPaginator:
"""Create a new pagination helper instance.

Args:
response: The HTTP ``requests.Response`` object.
previous_token: The previous page token value.
If the source API can make use of the `next_page_token_jsonpath`
attribute, or it contains a `X-Next-Page` header in the response
then you can remove this method.

If you need custom pagination that uses page numbers, "next" links, or
other approaches, please read the guide: https://sdk.meltano.com/en/v0.25.0/guides/pagination-classes.html.

Returns:
The next pagination token.
A pagination helper instance.
"""
# TODO: If pagination is required, return a token which can be used to get the
# next page. If this is the final page, return "None" to end the
# pagination loop.
if self.next_page_token_jsonpath:
all_matches = extract_jsonpath(
self.next_page_token_jsonpath,
response.json(),
)
first_match = next(iter(all_matches), None)
next_page_token = first_match
else:
next_page_token = response.headers.get("X-Next-Page", None)

return next_page_token
return super().get_new_paginator()

def get_url_params(
self,
Expand Down
6 changes: 3 additions & 3 deletions poetry.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.