API Polling & Rate Limiting

This module is part of the Data Ingestion & Grant Parsing Workflows reference, where it functions as the external acquisition boundary — the first stage that touches a funder’s network. Its mandate is strictly bounded: schedule polling against grant portal APIs, enforce rate-limit compliance, deduplicate responses, and emit validated raw payloads with a complete audit trail. It terminates at the validation gate and dispatches downstream.

Explicitly out of scope. No semantic interpretation, budget reconciliation, or compliance-rule evaluation happens here. Document-level extraction belongs to PDF Grant Application Parsing; spreadsheet validation belongs to Excel Budget Template Sync; field-level canonicalization belongs to Field Mapping & Normalization; transport failures and triage routing belong to Error Categorization & Logging; concurrent execution downstream is governed by Async Batch Processing Pipelines. The operational boundary ends precisely at successful HTTP response capture, payload schema validation, and structured audit logging. Any logic beyond transport verification violates the separation of concerns and introduces non-deterministic behavior into the pipeline.

The four stakeholders served are nonprofit operations teams and grant managers (predictable, traceable sync windows), Python automation developers (deterministic transport contracts and structured error returns), and compliance officers (immutable rate-limit and deduplication evidence).

Prerequisites #

This module pins its runtime so that polling windows and idempotency keys are reproducible across CI, staging, and multi-node production workers.

• Python: 3.11 or newer (relies on datetime.UTC semantics and asyncio.TaskGroup).
• Pinned packages:

# requirements.txt (acquisition boundary only)
httpx==0.27.0           # async transport, HTTP/2, connection pooling
tenacity==8.3.0         # declarative retry / backoff policies
redis==5.0.4            # distributed checksum + idempotency state store
orjson==3.10.3          # fast, deterministic JSON serialization
pytest==8.2.0           # transport-boundary tests
pytest-asyncio==0.23.7  # async test execution
respx==0.21.1           # httpx request mocking for rate-limit scenarios

• Environment variables: PORTAL_BASE_URL (funder API root), PORTAL_AUTH_TOKEN (Bearer credential, sourced from a secrets manager — never committed), POLL_WINDOW_SECONDS (scheduler cadence, e.g. 900 for 15-minute incremental syncs), RATE_REMAINING_FLOOR (proactive throttle threshold, default 5), and STATE_STORE_URL (Redis/Postgres DSN for the checksum registry).
• Upstream dependency: none — this is the network ingress. It assumes a deterministic scheduler (Celery Beat, a Kubernetes CronJob, or an asyncio loop) has already resolved which endpoints and windows are due. It does not decide what to poll; it executes a PollingContext it is handed.

Canonical polling configuration #

Each polling job instantiates one canonical async HTTP client and attaches four mandatory transport headers before transmission: Accept, Authorization, Idempotency-Key, and X-Client-Trace-ID. The Idempotency-Key is derived from a SHA-256 hash of the endpoint path, the sorted query parameters, and the scheduled window timestamp. This guarantees that duplicate network retries yield identical transport responses without triggering redundant downstream processing.

Polling intervals are configured per portal specification — typically 15-minute incremental syncs through 24-hour full reconciliation pulls. Developers must enforce strict timeout boundaries (default 30 s connect, 45 s read) to prevent thread starvation during portal degradation, and bound the connection pool so a single slow funder cannot exhaust workers shared across tenants.

Core implementation #

The reference implementation below demonstrates production-ready polling with explicit audit hooks, deterministic idempotency, and strict boundary enforcement. It uses httpx for async transport, the standard-library logger for structured compliance logging, and a pluggable state backend for deduplication. Every exit path returns a structured value or None — no exception is swallowed silently, and no raw payload is parsed inside this class.

import asyncio
import hashlib
import logging
import time
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import Dict, Optional

import httpx

# Structured audit logger compliant with SOC 2 CC6.1 / NIST SP 800-53 AU-2
AUDIT_LOGGER = logging.getLogger("grant_ingestion.transport_audit")


@dataclass
class PollingContext:
    """Deterministic execution context for transport-layer polling."""
    endpoint: str
    params: Dict[str, str]
    window_ts: datetime
    trace_id: str
    idempotency_key: str = field(init=False)

    def __post_init__(self) -> None:
        raw = f"{self.endpoint}|{sorted(self.params.items())}|{self.window_ts.isoformat()}"
        self.idempotency_key = hashlib.sha256(raw.encode("utf-8")).hexdigest()


class GrantPortalPoller:
    """
    Strict acquisition boundary for grant portal APIs.
    Terminates at payload validation. No semantic processing occurs here.
    """

    def __init__(
        self,
        base_url: str,
        auth_token: str,
        rate_limit_threshold: int = 5,
        timeout: float = 30.0,
    ) -> None:
        self.client = httpx.AsyncClient(
            base_url=base_url,
            http2=True,
            timeout=httpx.Timeout(timeout, read=45.0),
            limits=httpx.Limits(max_connections=10, max_keepalive_connections=5),
        )
        self.auth_token = auth_token
        self.rate_limit_threshold = rate_limit_threshold
        # In production, replace this dict with a Redis / PostgreSQL state store.
        self._checksum_registry: Dict[str, str] = {}

    def _build_headers(self, ctx: PollingContext) -> Dict[str, str]:
        return {
            "Accept": "application/json",
            "Authorization": f"Bearer {self.auth_token}",
            "Idempotency-Key": ctx.idempotency_key,
            "X-Client-Trace-ID": ctx.trace_id,
            "User-Agent": "GrantAutomation/2.1 (Compliance-Enforced)",
        }

    def _parse_rate_limits(self, headers: httpx.Headers) -> Dict[str, int]:
        return {
            "limit": int(headers.get("X-RateLimit-Limit", 0)),
            "remaining": int(headers.get("X-RateLimit-Remaining", 0)),
            "reset": int(headers.get("X-RateLimit-Reset", 0)),
            "retry_after": int(headers.get("Retry-After", 0)),
        }

    async def execute_poll(self, ctx: PollingContext) -> Optional[bytes]:
        headers = self._build_headers(ctx)
        response = await self.client.get(ctx.endpoint, params=ctx.params, headers=headers)
        limits = self._parse_rate_limits(response.headers)

        # AUDIT HOOK: transport capture (SOC 2 CC6.1 / NIST AU-2)
        AUDIT_LOGGER.info(
            "TRANSPORT_CAPTURE",
            extra={
                "trace_id": ctx.trace_id,
                "status": response.status_code,
                "rate_remaining": limits["remaining"],
                "idempotency_key": ctx.idempotency_key,
                "timestamp_utc": datetime.now(timezone.utc).isoformat(),
            },
        )

        if response.status_code == 429:
            await self._handle_rate_limit(response, limits, ctx.trace_id)
            return None

        if response.status_code != 200:
            # Structured error return — routed to Error Categorization & Logging.
            AUDIT_LOGGER.error(
                "TRANSPORT_FAILURE",
                extra={"trace_id": ctx.trace_id, "status": response.status_code},
            )
            return None

        payload = response.content
        current_checksum = hashlib.sha256(payload).hexdigest()

        # DEDUPLICATION CHECK (2 CFR 200 data integrity)
        if self._checksum_registry.get(ctx.idempotency_key) == current_checksum:
            AUDIT_LOGGER.info(
                "NO_CHANGE_DETECTED",
                extra={"trace_id": ctx.trace_id, "checksum": current_checksum},
            )
            return None

        self._checksum_registry[ctx.idempotency_key] = current_checksum
        AUDIT_LOGGER.info(
            "PAYLOAD_VALIDATED",
            extra={
                "trace_id": ctx.trace_id,
                "checksum": current_checksum,
                "size_bytes": len(payload),
            },
        )

        # Boundary enforcement: return raw bytes only.
        # Routing to downstream parsers occurs outside this class.
        return payload

    async def _handle_rate_limit(
        self, response: httpx.Response, limits: Dict[str, int], trace_id: str
    ) -> None:
        wait_time = limits["retry_after"] or max(0, limits["reset"] - int(time.time()))
        AUDIT_LOGGER.warning(
            "RATE_LIMIT_ENFORCED",
            extra={"wait_seconds": wait_time, "trace_id": trace_id},
        )
        await asyncio.sleep(wait_time)

    async def close(self) -> None:
        await self.client.aclose()

Three contracts make this class auditable. First, execute_poll returns either novel raw bytes or None; callers never receive a parsed object, so semantic logic cannot leak across the boundary. Second, every status branch emits a typed audit event keyed by trace_id. Third, deduplication is keyed by the deterministic idempotency_key, so a restarted scheduler that re-polls the same window produces NO_CHANGE_DETECTED rather than a duplicate downstream dispatch.

Rate-limit header contract #

Grant portals advertise their throttling state through headers, but the names drift between funders. The acquisition layer normalizes these aliases into one canonical rate-limit record before any backoff decision is made. The canonical contract follows RFC 6585 (the 429 Too Many Requests and Retry-After semantics).

When remaining drops below RATE_REMAINING_FLOOR, the scheduler proactively throttles the rest of the execution window rather than waiting for a hard 429. When a 429 does arrive, the pipeline waits the exact retry_after (or the delta to reset), never a guessed constant. Portal-specific header quirks, token-bucket reconstruction, and exponential-backoff-with-jitter strategies are worked end to end in Handling rate limits in grant portal APIs.

Deduplication & state contract #

On a 200 OK, the module computes a SHA-256 checksum of the raw response body and compares it against the state store keyed by idempotency_key. A match logs NO_CHANGE and halts the cycle; a novel checksum is persisted and the payload proceeds. The state store is the single source of truth for “have we already ingested this exact window,” and it must outlive any single worker.

Validation & testing #

Transport boundaries are tested without touching a live funder. respx mocks httpx so each rate-limit and deduplication branch is asserted deterministically, and audit-log emission is verified with caplog.

import asyncio
from datetime import datetime, timezone

import httpx
import pytest
import respx

from poller import GrantPortalPoller, PollingContext


def _ctx() -> PollingContext:
    return PollingContext(
        endpoint="/grants",
        params={"status": "open"},
        window_ts=datetime(2026, 6, 27, tzinfo=timezone.utc),
        trace_id="trace-001",
    )


@pytest.mark.asyncio
@respx.mock
async def test_novel_payload_is_returned_and_then_deduplicated(caplog):
    body = b'{"grants": [1, 2, 3]}'
    respx.get("https://portal.example/grants").mock(
        return_value=httpx.Response(200, content=body, headers={"X-RateLimit-Remaining": "42"})
    )
    poller = GrantPortalPoller("https://portal.example", "tok", timeout=5.0)
    ctx = _ctx()

    first = await poller.execute_poll(ctx)
    second = await poller.execute_poll(ctx)  # identical window → dedup

    assert first == body                  # novel payload crosses the boundary
    assert second is None                 # second pass is suppressed
    assert "PAYLOAD_VALIDATED" in caplog.text
    assert "NO_CHANGE_DETECTED" in caplog.text
    await poller.close()


@pytest.mark.asyncio
@respx.mock
async def test_429_waits_exact_retry_after(monkeypatch):
    respx.get("https://portal.example/grants").mock(
        return_value=httpx.Response(429, headers={"Retry-After": "7"})
    )
    slept: list[float] = []

    async def _fake_sleep(seconds: float) -> None:
        slept.append(seconds)

    monkeypatch.setattr(asyncio, "sleep", _fake_sleep)
    poller = GrantPortalPoller("https://portal.example", "tok")

    result = await poller.execute_poll(_ctx())

    assert result is None        # throttled cycles never emit a payload
    assert slept == [7]          # exact Retry-After honored, no guessing
    await poller.close()

The two expectations that matter for audit readiness: a throttled cycle must return None (never a partial payload) and must wait the server-stated interval; a re-polled identical window must produce exactly one PAYLOAD_VALIDATED followed by NO_CHANGE_DETECTED. Property-based tests with hypothesis can fuzz header casing and missing-header permutations against _parse_rate_limits to prove the coercion rules never raise.

Performance & scale considerations #

Nonprofit-scale workloads are bursty but small: a few dozen funder endpoints polled on staggered windows, not millions of requests per second. Tune for predictability, not raw throughput.

• Connection pooling: cap max_connections per portal (default 10) so one degraded funder cannot starve workers shared across tenants. Enable HTTP/2 to multiplex incremental syncs over a single connection.
• Window staggering: jitter scheduled windows by a few seconds so all endpoints do not fire on the same minute boundary and self-induce a 429.
• State-store sizing: the checksum registry stores one short hash per (endpoint, window) pair. Even with 24-hour retention across hundreds of endpoints, footprint stays in the low megabytes — set a TTL equal to your longest reconciliation window plus a margin.
• Backpressure: validated payloads are handed off to Async Batch Processing Pipelines through a bounded queue. If that queue is full, the poller should defer rather than buffer unbounded bytes in memory.
• Memory ceiling: because execute_poll returns raw bytes and immediately dispatches, peak memory is bounded by the largest single response, not by the batch — keep large attachment endpoints on their own concurrency budget.

Failure modes & troubleshooting #

Persistent or unclassifiable transport failures are not retried in place — they are forwarded to Error Categorization & Logging for triage, and recoverable network faults follow the shared retry contract in Pipeline Fallback & Retry Logic.

Compliance alignment #

The acquisition layer produces the transport-level evidence later stages depend on. Each control maps to concrete log fields, not generic assurances.

Credential material must never be persisted in logs or the checksum registry. Every transport log must carry X-Client-Trace-ID, a UTC ISO-8601 timestamp, the rate-limit header snapshot, and the deduplication outcome. Downstream, these trace_id-keyed records are correlated with the canonical artifacts emitted by the Core Architecture & Compliance Mapping reference and protected under Data Security & Access Boundaries.

Related #

• Parent reference: Data Ingestion & Grant Parsing Workflows
• Handling rate limits in grant portal APIs — backoff, jitter, and circuit-breaker patterns
• Async Batch Processing Pipelines — concurrent execution of validated payloads
• Error Categorization & Logging — triage routing for transport failures
• Field Mapping & Normalization — canonicalization of the payloads this stage acquires
• Cross-reference: Pipeline Fallback & Retry Logic