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: Any, **kwargs: Any) None

Create a new paginator.

Parameters
  • args – Paginator positional arguments for base class.

  • kwargs – Paginator keyword arguments for base class.

get_next(response: Response) ParseResult | None

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

Parameters

response – API response object.

Returns

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

abstract get_next_url(response: Response) str | None

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

Parameters

response – API response object.