Migrate your REST tap to use the new pagination classes!#
This guide will walk you through the process of migrating your REST tap to use the new pagination classes.
What’s changed?#
In a future release of the SDK, the RESTStream.get_next_page_token
method will no longer be
called as part of the stream pagination process. It is replaced by the
RESTStream.get_new_paginator
.
In the meantime, the get_next_page_token
method will still be called if you define it
in your stream class, but a deprecation warning will be logged, if
warnings are enabled.
Note
If you have not overridden the get_next_page_token
method in your stream class, you do not need
to make any changes.
Why is this change being made?#
The new pagination classes are designed to be more flexible, easier to understand, and easier to test. They also align better with the Single Responsibility Principle.
How to migrate#
Example: HATEOAS pagination, a.k.a. “next” links#
A common pattern in REST APIs is to use a next
field in the response to indicate the
URL of the next page of results. The BaseHATEOASPaginator
class can be used to handle this pattern.
# Original implementation
from urllib.parse import parse_qsl
class MyStream(RESTStream):
def get_next_page_token(self, response, previous_token):
data = response.json()
return data.get("next")
def get_url_params(self, context, next_page_token):
params = {}
if next_page_token:
params.update(parse_qsl(next_page_token.query))
return params
# New implementation
from singer_sdk.pagination import BaseHATEOASPaginator
class MyPaginator(BaseHATEOASPaginator):
def get_next_url(self, response):
data = response.json()
return data.get("next")
class MyStream(RESTStream):
def get_new_paginator(self):
return MyPaginator()
def get_url_params(self, context, next_page_token):
params = {}
# Next page token is a URL, so we can to parse it to extract the query string
if next_page_token:
params.update(parse_qsl(next_page_token.query))
return params
Example: Offset pagination#
Another common pattern is to use an offset
parameter to indicate the starting point of the next
page of results. The BaseOffsetPaginator
class can be used to handle this pattern.
# New implementation
from singer_sdk.pagination import BaseOffsetPaginator
class MyPaginator(BaseOffsetPaginator):
def has_more(self, response):
data = response.json()
return data.get("has_more", False)
class MyStream(RESTStream):
def get_new_paginator(self):
return BaseOffsetPaginator(start_value=0, page_size=250)
def get_url_params(self, context, next_page_token):
params = {}
# Next page token is an offset
if next_page_token:
params["offset"] = next_page_token
return params