The API i'm pulling from for my tap doesn't suppor...
# singer-tap-development
t
The API i'm pulling from for my tap doesn't support pagination. I'm having a hard time finding documentation on how to disable that. I can see when I run my tap that it's making a handful of calls to my API. The docs for
request_records())
on
RESTStream
say: "If pagination is detected, pages will be recursed automatically." but I'm not seeing how it's detecting pagination in this case
1
a
Is your API public? Could you share a link to any docs? When you say 'doesn't support pagination', does a single request to
tickets
(for example) return all the tickets? If you are building a custom tap, the docs on pagination might help https://sdk.meltano.com/en/v0.43.1/guides/pagination-classes.html
t
Yes, a single request to
/devices
returns all devices. I'm not able to find the docs publicly. They seems to only provide them if you have your own instance of their software. Now that we've clearedd up that pagination is not supported by this API I would like to have my custom tap not try multiple times. I'll read through those docs and hopefully find a solution
I've read that page. It doesn't talk about this issue I'm running into
a
Assuming you're using the SDK, Is your base stream inheriting from
RESTStream
? In which can you could make your method
get_new_paginator
return an instance of
SinglePagePaginator
https://sdk.meltano.com/en/v0.43.1/guides/pagination-classes.html https://sdk.meltano.com/en/v0.43.1/classes/singer_sdk.RESTStream.html#singer_sdk.RESTStream.get_new_paginator https://sdk.meltano.com/en/v0.43.1/classes/singer_sdk.pagination.SinglePagePaginator.html Just to confirm, you aren't implementing a child stream from this stream? That might explain multiple calls. Any code you can share here for your custom tap would help
t
I am doing a child stream. That could explain it. Here's what I have so far
Copy code
def get_child_context(self, record, context):
        """Return a context dictionary for child streams."""
        return {
            "device-name": record["device-name"],
        }
Here's client.py
Copy code
from __future__ import annotations
import decimal
import typing as t
from importlib import resources
from singer_sdk.authenticators import APIKeyAuthenticator
from singer_sdk.helpers.jsonpath import extract_jsonpath
from singer_sdk.pagination import BaseAPIPaginator  # noqa: TC002
from singer_sdk.streams import RESTStream

if t.TYPE_CHECKING:
    import requests
    from singer_sdk.helpers.types import Context

# TODO: Delete this is if not using json files for schema definition
SCHEMAS_DIR = resources.files(__package__) / "schemas"

class UispStream(RESTStream):
    records_jsonpath = "$[*]"

    @property
    def url_base(self) -> str:
        return "<https://uisp.utbb.net/nms/api/v2.1>"

    @property
    def authenticator(self) -> APIKeyAuthenticator:
        return APIKeyAuthenticator.create_for_stream(
            self,
            key="x-auth-token",
            # value=self.config.get("auth_token", ""),
            value="thetoken",
            location="header",
        )

    def parse_response(self, response: requests.Response) -> t.Iterable[dict]:
        records = extract_jsonpath(
            self.records_jsonpath,
            input=response.json(parse_float=decimal.Decimal),
        )
        for record in records:
            identification = record.get("identification")
            if isinstance(identification, dict) and "id" in identification:
                record["id"] = identification["id"]
                record["mac"] = identification["mac"]
                record.pop("identification", None)
            yield record

    def post_process(
        self,
        row: dict,
        context: Context | None = None,  # noqa: ARG002
    ) -> dict | None:
        """As needed, append or transform raw data to match expected structure.

        Args:
            row: An individual record from the stream.
            context: The stream context.

        Returns:
            The updated record dictionary, or ``None`` to skip the record.
        """
        # TODO: Delete this method if not needed.
        return row
If I'm understanding the parent child implementation correctly then there will be one call to the API for every child, is that right? If I were doing this just in a script I would get all parents with one call, all the children with another, then loop over the parents to create my a dictionary like this
Copy code
{
  id: the_child_id,
  name: child
  parent_id: the_parent_id
}
Is it possible to get all the children with one call with the tap or do I need to do one call per child?
Figured it out
👍 1