Hanroux Vos
07/10/2024, 2:51 PM"""Stream type classes for tap-dogsapi."""
from __future__ import annotations
import sys
import typing as t
from singer_sdk import typing as th # type: ignore # JSON Schema typing helpers
from tap_dogsapi.client import DogsAPIStream
if sys.version_info >= (3, 9):
import importlib.resources as importlib_resources
else:
import importlib_resources
class BreedStream(DogsAPIStream):
"""Define custom stream."""
name = "breeds"
path = "/breeds"
primary_keys: t.ClassVar[list[str]] = ["id"]
replication_key = None
# Optionally, you may also use `schema_filepath` in place of `schema`:
# schema_filepath = SCHEMAS_DIR / "users.json" # noqa: ERA001
schema = th.PropertiesList(
th.Property(
"id",
th.StringType,
description="The breed's system ID",
),
th.Property(
"breed",
th.StringType,
description="The breed of the dog",
),
th.Property(
"attributes",
th.ObjectType(
th.Property("name",th.StringType),
th.Property("description", th.StringType),
th.Property("hypoallergenic",th.BooleanType),
)
),
).to_dict()
Hanroux Vos
07/10/2024, 2:53 PMclass DogsAPIStream(RESTStream):
"""DogsAPI stream class."""
@property
def url_base(self) -> str:
"""Return the API URL root, configurable via tap settings."""
return "<https://dogapi.dog/api/v2>"
Edgar Ramírez (Arch.dev)
07/10/2024, 3:05 PMrecords_jsonpath
in your client classHanroux Vos
07/10/2024, 3:46 PMrecords_jsonpath = "$['data'][*]"
visch
07/10/2024, 3:52 PM