Is there a some convention for where custom paginator classes should be defined in an SDK tap? I have a couple sitting in

client.py

and

streams.py

currently, but wondering if they would fit better in a separate `paginators.py`/`pagination.py` module...

No hard-and-fast rule here. Moving to a separate page like

pagination.py

probably would make sense. The

client.py

file was always intended to handle the 'shared nuances' of a specific API, so that would be a fine place also if it isn't already too crowded there.

Thanks @aaronsteers. The reason I have some in

streams.py

is I ended up having to create a paginator class for a specific stream. As a separate thought, is this a valid approach or more of an anti-pattern?

pagination.py

class Auth0Paginator(HeaderLinkPaginator):
    def get_next_url(self, response: Response):
        url = super().get_next_url(response)
        return url if url and response.json() else None


class LogsPaginator(Auth0Paginator):
    def __init__(
        self,
        log_expired_callback: Callable[[Response], bool],
        *args,
        **kwargs,
    ):
        super().__init__(*args, **kwargs)
        self.log_expired_callback = log_expired_callback

    def get_next(self, response: Response):
        return (
            self.log_expired_callback(response)
            or super().get_next(response)
            or JSONPathPaginator("$[*].log_id").get_next(response)
        )

client.py

class Auth0Stream(RESTStream):
    def get_new_paginator(self):
        return Auth0Paginator()

streams.py

class LogsStream(Auth0Stream):
    def get_new_paginator(self):
        def log_expired_callback(response: requests.Response):
            self.log_expired = response.status_code == 400
            return self.log_expired

        return LogsPaginator(log_expired_callback)

(for context, I'm updating

tap-auth0

to the newest SDK version, and in the process implementing pagination classes in place of the deprecated

get_next_page_token

)

That’s a good approach!

log_expired_callback

could be a method of

LogsPaginator

too

How so? A paginator doesn't have access to the stream its instantiated for, right?

@Reuben (Matatika) yeah,

LogsStream.log_expired

is no longer used in any condition, so the class even not have a method to and just check the status:

class LogsPaginator(Auth0Paginator):
    def get_next(self, response: Response):
        return (
            response.status_code == 400
            or super().get_next(response)
            or JSONPathPaginator("$[*].log_id").get_next(response)
        )