0

Fetch all pages until an empty batch

by Frontier Interview· about 3 hours ago· ⑂ 0 remixes

Question: Fetch all pages until an empty batch

Companies: Sierra

Difficulty: Easy

Problem statement

Page through a batched API until it returns an empty batch, collecting every record without fetching forever on errors. Reported on 1point3acres' Sierra bank; reconstructed in our own words.

Clarify the setup

An API returns records in batches: fetch(page) gives a list, and an empty list means there is nothing more. Collect every record. The loop is three lines; the interview is about what goes wrong. Ask:

  • Page numbers or cursors? With page numbers (page=0, 1, 2…) you can fetch pages in parallel. With cursors (each response carries next_cursor) you cannot, because you only learn the next cursor from the previous response.
  • Which errors are transient? Timeouts, HTTP 429 and 5xx: retry with backoff. A 400 or 401 will not fix itself: fail fast.
  • What stops a runaway loop? An API bug that never returns an empty page, or returns the same cursor forever. Cap the pages and detect repeated cursors.
  • Return everything or stream? If there may be millions of records, yield them as they arrive instead of building a list.

Approach

The core loop. Fetch page after page, yielding records, until a page is empty.

Retries with exponential backoff and jitter. On a transient error, wait base × 2^attempt seconds, multiplied by a random factor between 0.5 and 1 so many clients do not retry in lockstep, capped at max_delay. Honor a server's Retry-After when given. After max_retries, raise with the page number, so the caller knows exactly how far it got and can resume. Pass sleep and the random source in as parameters: tests run instantly and deterministically.

Guards. Stop with an error after max_pages. In cursor mode, keep the set of cursors seen; a repeat means the API is looping.

Part 2: parallel pages. With page numbers, fetch a window of w pages at once with a thread pool. Yield results in page order. When any page in the window comes back empty, stop launching new pages, and drop the pages after the first empty one (they should be empty too, but if the data grew meanwhile they could hold records past the end you already chose). With w = 8 and 200 ms per call, 1,000 pages take about 1000 / 8 × 0.2 = 25 s instead of 200 s.

Code

import random
import time
from concurrent.futures import ThreadPoolExecutor
from typing import Any, Callable, Iterator, List, Optional, Tuple


class TransientError(Exception):
    def __init__(self, msg: str = "", retry_after: Optional[float] = None) -> None:
        super().__init__(msg)
        self.retry_after = retry_after


class FetchFailed(Exception):
    pass


def with_retries(
    call: Callable[[], Any],
    what: str,
    max_retries: int = 5,
    base_delay: float = 0.5,
    max_delay: float = 30.0,
    sleep: Callable[[float], None] = time.sleep,
    rng: random.Random = random.Random(),
) -> Any:
    for attempt in range(max_retries + 1):
        try:
            return call()
        except TransientError as e:
            if attempt == max_retries:
                raise FetchFailed(f"{what}: gave up after {max_retries} retries") from e
            delay = e.retry_after
            if delay is None:
                delay = min(max_delay, base_delay * 2 ** attempt) * (0.5 + rng.random() / 2)
            sleep(delay)


def fetch_all_pages(
    fetch: Callable[[int], List[Any]],
    max_pages: int = 10_000,
    **retry: Any,
) -> Iterator[Any]:
    """Page-number API: yield every record until the first empty page."""
    for page in range(max_pages):
        batch = with_retries(lambda: fetch(page), f"page {page}", **retry)
        if not batch:
            return
        yield from batch
    raise FetchFailed(f"no empty page within {max_pages} pages")


def fetch_all_cursor(
    fetch: Callable[[Optional[str]], Tuple[List[Any], Optional[str]]],
    max_pages: int = 10_000,
    **retry: Any,
) -> Iterator[Any]:
    """Cursor API: fetch(cursor) -> (records, next_cursor)."""
    cursor: Optional[str] = None
    seen = set()
    for _ in range(max_pages):
        records, nxt = with_retries(lambda: fetch(cursor), f"cursor {cursor!r}", **retry)
        yield from records
        if not records or nxt is None:
            return
        if nxt in seen:
            raise FetchFailed(f"cursor {nxt!r} repeated: the API is looping")
        seen.add(nxt)
        cursor = nxt
    raise FetchFailed(f"no end within {max_pages} pages")


def fetch_all_parallel(
    fetch: Callable[[int], List[Any]],
    window: int = 8,
    max_pages: int = 10_000,
    **retry: Any,
) -> List[Any]:
    """Part 2: fetch `window` pages at a time; results in page order."""
    out: List[Any] = []
    with ThreadPoolExecutor(max_workers=window) as pool:
        for first in range(0, max_pages, window):
            pages = range(first, min(first + window, max_pages))
            futures = [pool.submit(with_retries, (lambda p=p: fetch(p)), f"page {p}", **retry) for p in pages]
            for fut in futures:                  # in page order
                batch = fut.result()
                if not batch:
                    for rest in futures:
                        rest.cancel()
                    return out
                out.extend(batch)
    raise FetchFailed(f"no empty page within {max_pages} pages")

The lambdas in fetch_all_pages and fetch_all_cursor are called immediately inside with_retries, so they see the current page or cursor. In the parallel version they run later, which is why p=p binds each page number.

Complexity

  • P pages and R records: O(P) calls and O(R) work. Memory: one page at a time for the generators, all records for the parallel list.
  • Worst-case added latency from retries per page: the sum of the backoff delays, 0.5 + 1 + 2 + 4 + 8 = 15.5 s for 5 retries before jitter. Set max_retries from how long the caller can afford to wait.

Test cases

class FakeAPI:
    def __init__(self, records, page_size=3, failures=None):
        self.pages = [records[i:i + page_size] for i in range(0, len(records), page_size)]
        self.failures = dict(failures or {})     # page -> how many times to fail first
        self.calls = []

    def fetch(self, page):
        self.calls.append(page)
        if self.failures.get(page, 0) > 0:
            self.failures[page] -= 1
            raise TransientError("503")
        return list(self.pages[page]) if page < len(self.pages) else []


def _no_sleep(_):
    pass


def test_collects_everything_and_stops_at_empty_page():
    api = FakeAPI(list(range(10)))
    assert list(fetch_all_pages(api.fetch, sleep=_no_sleep)) == list(range(10))
    assert api.calls == [0, 1, 2, 3, 4]
    assert list(fetch_all_pages(FakeAPI([]).fetch, sleep=_no_sleep)) == []


def test_retries_transient_errors_with_backoff():
    api = FakeAPI(list(range(7)), failures={1: 2})
    delays = []
    out = list(fetch_all_pages(api.fetch, sleep=delays.append, rng=random.Random(0)))
    assert out == list(range(7)) and len(delays) == 2
    assert 0.25 <= delays[0] <= 0.5 and 0.5 <= delays[1] <= 1.0


def test_gives_up_and_names_the_page():
    api = FakeAPI(list(range(7)), failures={2: 99})
    got = []
    try:
        for r in fetch_all_pages(api.fetch, max_retries=3, sleep=_no_sleep):
            got.append(r)
        assert False
    except FetchFailed as e:
        assert "page 2" in str(e)
    assert got == list(range(6))                 # records before the failure were delivered


def test_runaway_api_is_stopped():
    try:
        list(fetch_all_pages(lambda p: [p], max_pages=50, sleep=_no_sleep))
        assert False
    except FetchFailed:
        pass

def test_cursor_mode_and_loop_detection():
    data = {None: ([1, 2], "a"), "a": ([3], "b"), "b": ([], None)}
    assert list(fetch_all_cursor(lambda c: data[c], sleep=_no_sleep)) == [1, 2, 3]
    looping = {None: ([1], "x"), "x": ([2], "y"), "y": ([3], "x")}
    try:
        list(fetch_all_cursor(lambda c: looping[c], sleep=_no_sleep))
        assert False
    except FetchFailed as e:
        assert "repeated" in str(e)


def test_retry_after_is_honored():
    calls = {"n": 0}

    def fetch(page):
        calls["n"] += 1
        if calls["n"] == 1:
            raise TransientError("429", retry_after=7.0)
        return [] if page else [1]

    delays = []
    assert list(fetch_all_pages(fetch, sleep=delays.append)) == [1]
    assert delays == [7.0]


def test_parallel_matches_sequential_in_order():
    import threading
    lock = threading.Lock()
    api = FakeAPI(list(range(100)), page_size=7, failures={3: 1, 9: 2})
    real_fetch = api.fetch

    def fetch(page):
        with lock:
            return real_fetch(page)

    assert fetch_all_parallel(fetch, window=4, sleep=_no_sleep) == list(range(100))

Follow-ups

  1. Resuming. For a long export, persist the last completed page or cursor after each batch. On restart, continue from there instead of from page 0. Make the consumer idempotent by record id, since the last batch may be processed twice.
  2. Consistency while paging. Offset-based pages shift if records are inserted or deleted during the crawl: you can skip or repeat records. Cursor pagination over a stable sort key (created_at, id) avoids it; otherwise de-duplicate by id.
  3. Rate limits. Parallel fetching can trigger 429s. Use a shared token bucket across workers and shrink the window when 429s appear (additive increase, multiplicative decrease).
  4. Async. With asyncio, a semaphore of size w around fetch replaces the thread pool, and thousands of concurrent requests cost no threads.
  5. For an agent tool. When an agent calls a paginated tool, return the first page plus a "more available" note and let the model ask for more, instead of dumping thousands of records into its context.

Interviewer expectations

Mid-level (L4)

  • A correct loop that stops on the empty page, with the edge case of an empty first page.
  • Retries on errors with some backoff.
  • Tests with a fake API.

Senior (L5)

  • Separates transient from permanent errors, uses exponential backoff with jitter and Retry-After, and fails with the page number after a bounded number of retries.
  • Guards against runaway APIs (page cap, repeated cursors) and streams records instead of buffering.
  • Injects sleep and randomness so the tests are instant and deterministic.

Staff (L6+)

  • Parallelizes page-number APIs with an ordered window, explains why cursor APIs cannot be parallelized that way, and adapts to rate limits.
  • Plans for resumable exports, offset drift and idempotent consumers.
  • Thinks about the caller: how much latency retries may add, and how an agent should consume a paginated tool.

Sign in to comment, vote or remix.