Hello All, I am very new to meltano. I am trying to build a REST tap . Although everything is easy to build. I am stuck at pagination. My REST api support pagination with POST request with

/query

endpoint. This is how pagination is supported. I am clueless about how to make this happen in my custom tap development.

import requests
import json

url = "<https://api.domain.com/v4/endpoint/query>"

payload = json.dumps({
  "options": {
    "page": 2
  }
})
headers = {
  'Content-Type': 'application/json'
}

response = requests.request("POST", url, headers=headers, data=payload)

print(response.text)

output of above request give back below details along with data

"totalDocs": 205,
    "limit": 10,
    "totalPages": 21,
    "page": 2,
    "hasPrevPage": true,
    "hasNextPage": true,
    "prevPage": 1,
    "nextPage": 3

FYI , to get data without pagination , i have to make /GET request without any payload

This is a fairly common type of pagination. So fortunately there are a few examples to take from. I implemented something very similar here: https://github.com/hholgersen/tap-dbtapi/blob/c121fa3e86d59c6917ee0497481202c778ad7393/tap_dbtapi/streams.py#L182. The get_params method is where you would return the json payload, the get_next_page_token is where you would grab the nextPage number from the output and return it.

Thanks @Henning Holgersen. Do i need to override the request type to POST

You are right, of course, I was a little quick in finding an example. The method you want to override is

prepare_request_payload

, but the principle is the same. And you need to specify

rest_method="POST"

in the stream class, like this: https://github.com/radbrt/tap-prefect/blob/85428f4443700fa53a49d5d5e072968f783ea441/tap_prefect/streams.py#L46 I see that class also has a

prepare_request_payload

method you can look at.

@mahesh_kalani you could also use a pagination class:

from singer_sdk.pagination import BasePageNumberPaginator

class MyAPIPaginator(BasePageNumberPaginator):
    def has_more(self, response):
        data = response.json()
        return data["totalDocs"]["hasNextPage"]

class MyStream(RESTStream):
    def get_new_paginator(self):
        return MyAPIPaginator(start_value=1)  # first page is 1

https://sdk.meltano.com/en/latest/reference.html#pagination

nice thanks