Hi There! I’m faily new to Meltano / Singer and developing my own tap for a rest API. The call I make returns a single object with data over a certain timeframe. The timeframe is queried with parameters. For example:

REQUEST:
GET <https://uri.example.com/resource?from_ts=1682380800&to_ts=1682985600>

RESPONSE:

{
"from_date": 1682380800000,
"to_date": 1682985600000,
"key": "value" 
}

The timeframe is limited for let’s say a week. What would be a good Meltano way to replicate data over a longer timeframe? I’m thinking of using the to_date as a replication key and use a paginator to keep track of the next calls. Does anyone know a good example on this? Any help is appreciated :)

@adam_tetz I have dealt with APIs like that in the past. My approach was to paginate through dates. Instead of using page numbers, you use the next start date and the page size is the window. In this case, the window is 7 days in seconds.

Thanks for the idea! @josue_sehnem I’ll give this a try :)

if you have trouble let me know, and I can try to find some taps that I developed and use that

The SDK has helpful pagination helper classes, and you could implement your own to iterate over a range of dates:

@dataclasses.dataclass
class Range(t.Generic[T]):
    """A range of values."""
    lower: T | None
    upper: T


class NumberRangePaginator(BaseAPIPaginator[Range[float]]):
    """Paginator that works with number ranges."""

    def __init__(
        self,
        start: float,
        end: float,
        step: float,
        *args: t.Any,
        **kwargs: t.Any,
    ) -> None:
        """Create a new paginator."""
        start_value = Range(None, start)
        super().__init__(start_value, *args, **kwargs)
        self._end = end

        if step <= 0:
            msg = "Step must be a positive number."
            raise ValueError(msg)

        self._step = step

    def get_next(self, response: Response) -> Range[float] | None:
        """Get the next page value.

        Args:
            response: API response object.

        Returns:
            The next page value.
        """
        next_value = self._value.upper + self._step
        if next_value >= self._end:
            return None

        return Range(self._value.upper, next_value)

Thank you Edgar!