Tanner Wilcox
02/27/2025, 11:38 PMrequest_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 caseAndy Carter
02/28/2025, 9:00 AMtickets
(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.htmlTanner Wilcox
02/28/2025, 3:53 PM/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 solutionTanner Wilcox
02/28/2025, 3:58 PMAndy Carter
02/28/2025, 5:06 PMRESTStream
? 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 helpTanner Wilcox
02/28/2025, 6:01 PMdef get_child_context(self, record, context):
"""Return a context dictionary for child streams."""
return {
"device-name": record["device-name"],
}
Here's client.py
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
{
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?Tanner Wilcox
02/28/2025, 9:31 PM