Paginating Grants.gov Search Results Deterministically

This guide is part of the API Polling & Rate Limiting section within the broader Data Ingestion & Grant Parsing Workflows framework, and it solves one narrow problem: how do you walk every page of a Grants.gov search2 opportunity query so that two runs of the same query return the identical, complete set — no skips, no duplicates, and no silent short-count?

The naive answer — loop startRecordNum from 0 in steps of rows until a page comes back short — passes on a quiet afternoon and corrupts your dataset the moment a funder posts a new opportunity mid-crawl. Offset pagination assumes the underlying result set is frozen while you page through it. It is not. Grants.gov re-ranks and re-counts on every request, so a single insertion shifts every downstream record by one position, and your fixed offsets skip one opportunity and re-read another. This guide builds a pager that fixes a stable sort key, walks the set by keyset cursor instead of raw offset, deduplicates by opportunityId, checkpoints the last-seen key so a crash resumes instead of restarting, and reconciles the accumulated count against the API’s reported hitCount — raising on mismatch rather than persisting a partial set.

When to Use This Approach #

Reach for a deterministic pager whenever all three of these hold:

• The crawl must be reproducible for audit. Federal award data ingested from Grants.gov feeds records that must be traceable under 2 CFR §200.302, which requires financial-management systems to produce records that identify the source and application of funds. A crawl that silently drops an opportunity because pagination skipped it is a data-integrity defect, not a cosmetic one — the reconciliation gate in Step 5 is what turns a silent short-count into a hard failure.
• The result set is large and mutating. Grants.gov re-ranks and re-counts on every search2 call. If your query returns more than one page and the corpus is actively updated — which the live opportunities feed always is — offset pagination will skip and duplicate records under you. Keyset pagination on a stable key (Steps 1–2) removes that window.
• The job may crash and must resume. A long crawl that dies at page 40 should resume at page 40, not restart at offset zero and re-download everything. The checkpoint in Step 4 persists the last-seen sort key so a restart is idempotent.

Request cadence, backoff, and quota accounting are explicitly out of scope here — that pacing belongs to Handling Rate Limits in Grant Portal APIs and the token-bucket shaping in Implementing Token-Bucket Throttling for Grants.gov APIs. Concurrent dispatch of the fetched pages belongs to Async Batch Processing Pipelines. This pager runs one page at a time, in order, so the cursor walk and the count reconciliation stay deterministic.

Step-by-Step Implementation #

The reference implementation targets Python 3.10+ and depends only on requests plus the standard library. Grants.gov’s search2 endpoint accepts a JSON body with keyword, rows, startRecordNum, and sortBy, and returns hitCount (the total matching the query) alongside an oppHits array.

requests==2.32.3

Step 1: Fix a stable sort key #

Determinism starts with ordering. If you do not pin sortBy, Grants.gov defaults to relevance, and relevance is recomputed per request — the same query can order the same records differently between two calls, which makes any positional pagination meaningless. Pin the sort to a monotonic, tie-broken key: openDate first, then opportunityId as the tiebreaker so two opportunities posted the same day still have one unambiguous order.

import logging
from dataclasses import dataclass, field
from typing import Optional

LOGGER = logging.getLogger("grants_gov.pagination")
LOGGER.setLevel(logging.INFO)
_handler = logging.StreamHandler()
_handler.setFormatter(logging.Formatter("%(asctime)s | %(levelname)s | %(message)s"))
LOGGER.addHandler(_handler)

SEARCH2_URL = "https://api.grants.gov/v1/api/search2"
# openDate|asc gives a monotonic primary key; opportunityId breaks same-day ties.
STABLE_SORT = "openDate|asc"


@dataclass(frozen=True)
class SortKey:
    """The composite key that fixes a total order over the result set."""
    open_date: str      # ISO date string, e.g. "2026-06-01"
    opportunity_id: str

    def as_tuple(self) -> tuple[str, str]:
        return (self.open_date, self.opportunity_id)

Making SortKey a frozen=True dataclass means it is hashable and comparable, so it can act as both a checkpoint value and a cursor bound. The open_date, opportunity_id tuple is the single source of ordering truth for every later step.

Step 2: Walk by keyset cursor, not raw offset #

Offset pagination asks the server for “records 200–249.” If any record before position 200 is inserted or removed between page 4 and page 5, position 200 now points at a record you already saw (a duplicate) or one past the one you needed (a skip). Keyset pagination instead asks “records after this sort key,” so an insertion earlier in the set cannot shift the window you have already passed. Grants.gov’s search2 does not expose a native cursor parameter, so we emulate keyset semantics: page in order and treat the last row’s SortKey as the high-water mark, discarding anything at or below a key we have already consumed.

import requests
from typing import Iterator


class PaginationTransportError(RuntimeError):
    """Raised when a page cannot be fetched or decoded."""


def fetch_page(keyword: str, start: int, rows: int, timeout: float = 30.0) -> dict:
    """Fetch one search2 page under the fixed stable sort. Pacing lives upstream."""
    payload = {"keyword": keyword, "rows": rows, "startRecordNum": start, "sortBy": STABLE_SORT}
    try:
        resp = requests.post(SEARCH2_URL, json=payload, timeout=timeout)
        resp.raise_for_status()
        body = resp.json()
    except (requests.RequestException, ValueError) as exc:
        LOGGER.error("search2 fetch failed at start=%d: %s", start, exc)
        raise PaginationTransportError(f"page at start={start} failed") from exc
    return body["data"]


def walk_pages(keyword: str, rows: int = 100) -> Iterator[tuple[list[dict], int]]:
    """Yield (page_hits, hit_count) walking the set in stable-sort order."""
    start = 0
    while True:
        data = fetch_page(keyword, start, rows)
        hits: list[dict] = data.get("oppHits", [])
        hit_count: int = int(data.get("hitCount", 0))
        if not hits:
            return
        yield hits, hit_count
        start += len(hits)

Fixing sortBy on every request is what makes the ascending startRecordNum walk equivalent to a keyset scan: because the order is stable, advancing start by the number of rows actually returned moves the high-water mark forward monotonically instead of jumping to a re-ranked position.

Step 3: Deduplicate by opportunityId across pages #

Even with a stable sort, a record inserted at the boundary between two pages can surface on both. The defense is idempotent accumulation: track every opportunityId you have emitted in a set and drop repeats. The set is the authority on “have I seen this,” never the page index.

def accumulate_unique(
    pages: Iterator[tuple[list[dict], int]],
    seen: set[str],
) -> Iterator[tuple[dict, int]]:
    """Emit each opportunity at most once, keyed on opportunityId."""
    for hits, hit_count in pages:
        for hit in hits:
            opp_id = str(hit["id"])
            if opp_id in seen:
                LOGGER.debug("skip duplicate opportunityId=%s across page boundary", opp_id)
                continue
            seen.add(opp_id)
            yield hit, hit_count

Threading seen in as a parameter rather than a local is deliberate: it is the same set the checkpoint restores in Step 4, so a resumed run does not re-emit opportunities it already persisted before the crash.

Step 4: Persist a resumable checkpoint #

A checkpoint is the last-seen SortKey plus the set of already-emitted ids, written to durable storage after each page. On restart you reload it, so the crawl resumes at the high-water mark instead of at offset zero. Write the checkpoint atomically (temp file plus os.replace) so a crash mid-write cannot leave a truncated checkpoint that corrupts the next resume.

import json
import os
from pathlib import Path


@dataclass
class Checkpoint:
    last_key: Optional[SortKey] = None
    seen_ids: set[str] = field(default_factory=set)

    def save(self, path: Path) -> None:
        """Atomically persist so a crash cannot leave a half-written checkpoint."""
        payload = {
            "last_key": self.last_key.as_tuple() if self.last_key else None,
            "seen_ids": sorted(self.seen_ids),
        }
        tmp = path.with_suffix(".tmp")
        tmp.write_text(json.dumps(payload))
        os.replace(tmp, path)  # atomic on POSIX
        LOGGER.info("checkpoint saved | seen=%d | last_key=%s", len(self.seen_ids), payload["last_key"])

    @classmethod
    def load(cls, path: Path) -> "Checkpoint":
        if not path.exists():
            return cls()
        raw = json.loads(path.read_text())
        key = SortKey(*raw["last_key"]) if raw["last_key"] else None
        return cls(last_key=key, seen_ids=set(raw["seen_ids"]))

Because seen_ids is restored alongside last_key, a resumed crawl replays only the current page’s boundary records and dedups them out — it never double-counts, which is what keeps the Step 5 reconciliation honest across restarts.

Step 5: Reconcile the accumulated count against the reported total #

The final gate is the one that earns the word “deterministic.” Every page carries hitCount, the total the API says matches the query. After the cursor is exhausted, compare your accumulated unique count to that total. An exact match persists the complete set; any mismatch raises rather than writing a partial result, so a truncated crawl can never masquerade as a full one downstream.

class PaginationIntegrityError(RuntimeError):
    """Raised when the accumulated unique count disagrees with the reported total."""


def crawl(keyword: str, checkpoint_path: Path, rows: int = 100) -> list[dict]:
    """Deterministic crawl: dedup, checkpoint, and reconcile against hitCount."""
    ckpt = Checkpoint.load(checkpoint_path)
    collected: list[dict] = []
    reported_total: Optional[int] = None

    for hit, hit_count in accumulate_unique(walk_pages(keyword, rows), ckpt.seen_ids):
        reported_total = hit_count
        collected.append(hit)
        ckpt.last_key = SortKey(hit.get("openDate", ""), str(hit["id"]))
        if len(collected) % rows == 0:
            ckpt.save(checkpoint_path)

    ckpt.save(checkpoint_path)
    unique_count = len(ckpt.seen_ids)
    if reported_total is not None and unique_count != reported_total:
        raise PaginationIntegrityError(
            f"reconciliation failed: collected {unique_count} unique ids "
            f"but hitCount reported {reported_total}"
        )
    LOGGER.info("crawl reconciled | unique=%d == hitCount=%s", unique_count, reported_total)
    return collected

The reconciliation reads hitCount from the last page rather than the first, because a mutating corpus can change the total mid-crawl; if the set grew while you paged, the mismatch is real and a re-crawl is the correct remediation, not a silent accept.

Verification #

Confirm the pager is deterministic with four checks:

1. Same query, same set. Run crawl twice against a stable query and assert the two returned id-sets are equal. This proves the fixed sortBy plus keyset walk produce a reproducible result rather than a relevance-ordered one.
2. Boundary insertions do not duplicate. Feed accumulate_unique two pages that share one opportunityId and assert the shared id is emitted exactly once and appears in seen.
3. A crash resumes, not restarts. Save a checkpoint after page 2, construct a fresh Checkpoint.load, and assert seen_ids is repopulated so re-emitted boundary records are deduped out.
4. A short count raises. Force hitCount above the number of unique ids collected and assert crawl raises PaginationIntegrityError and returns no set.

A compliant run emits one INFO reconciliation line naming the unique count and the hitCount they matched; ship those logs to a write-once tier so the crawl is replayable for the audit trail expected under 2 CFR §200.302.

from pathlib import Path

ckpt_path = Path("grants_crawl.ckpt")
first = {h["id"] for h in crawl("water quality", ckpt_path)}
ckpt_path.unlink(missing_ok=True)
second = {h["id"] for h in crawl("water quality", ckpt_path)}
assert first == second, "crawl is not reproducible across runs"

Common Errors & Fixes #

Related #

• Parent section: API Polling & Rate Limiting
• The pacing layer this pager assumes upstream: Handling Rate Limits in Grant Portal APIs
• Shaping request bursts against Grants.gov quotas: Implementing Token-Bucket Throttling for Grants.gov APIs
• Where the fetched pages get dispatched concurrently: Async Batch Processing Pipelines