singer_sdk.pagination.BaseHATEOASPaginator

class singer_sdk.pagination.BaseHATEOASPaginator

Paginator class for APIs supporting HATEOAS links in their response bodies.

HATEOAS stands for “Hypermedia as the Engine of Application State”. See https://en.wikipedia.org/wiki/HATEOAS.

This paginator expects responses to have a key “next” with a value like “https://api.com/link/to/next-item”.

The current_value attribute of this paginator is a urllib.parse.ParseResult object. This object contains the following attributes:

• scheme

• netloc

• path

• params

• query

• fragment

That means you can access and parse the query params in your stream like this:

class MyHATEOASPaginator(BaseHATEOASPaginator):
    def get_next_url(self, response):
        return response.json().get("next")


class MyStream(Stream):
    def get_new_paginator(self):
        return MyHATEOASPaginator()

    def get_url_params(self, next_page_token) -> dict:
        if next_page_token:
            return dict(parse_qsl(next_page_token.query))
        return {}

__init__(*args, **kwargs)

Create a new paginator.

Parameters:

• args (Any) – Paginator positional arguments for base class.

• kwargs (Any) – Paginator keyword arguments for base class.

Return type:

None

get_next(response)

Get the next pagination token or index from the API response.

Parameters:

response (requests.Response) – API response object.

Returns:

A parsed HATEOAS link if the response has one, otherwise None.

Return type:

ParseResult | None

abstractmethod get_next_url(response)

Override this method to extract a HATEOAS link from the response.

Parameters:

response (requests.Response) – API response object.

Return type:

str | None