Research Grant Automation Lab & Compliance Toolkit

Automating daily grant portal polling with Python requests

On this page

Problem statement

You need a once-a-day, unattended Python job that pulls award and status records from an external university grant portal, writes them to disk only when they actually change, and produces an immutable audit trail that satisfies federal sponsor data-integrity rules — without a manual download ever entering the chain of custody.

This task sits under API Polling & Portal Integration, part of the broader Automated Ingestion & Data Sync Workflows practice. The polling layer is intentionally narrow: it acquires data deterministically and hands it to the Schema Validation Pipelines and downstream archival layers. It does not interpret compliance state — it captures it, fingerprints it, and surfaces anomalies for human review, consistent with the separation of concerns established in the Grant Lifecycle Architecture Design.

Prerequisites

Before deploying the poller, confirm the following environment and policy configuration:

  • Python 3.10+ (the code uses from __future__-free union syntax and datetime.timezone.utc).
  • Libraries: requests (HTTP), urllib3 (bundled with requests, used for the retry adapter). Install with pip install "requests>=2.31".
  • Environment variables (never hard-code credentials, per Security Boundary Configuration):
    • GRANT_PORTAL_BASE_URL — the portal API root, e.g. https://api.university-funding.example/v1.
    • GRANT_PORTAL_API_TOKEN — a least-privilege, read-only bearer token scoped to grant-status endpoints only.
  • Policy config: a frozen institutional schema definition (the required-key set below) version-controlled alongside your University Policy Mapping Frameworks, plus a writable state directory (.compliance_state/) retained per your sponsor’s record-retention schedule (typically 3–7 years for federal awards).
  • Scheduler: cron or a systemd timer to fire the job once per day. The script is idempotent, so an accidental double-run causes no duplicate writes.

Step-by-step implementation

The flow below is enforced by the poller: a daily trigger fetches the endpoint, retries transient failures with backoff, validates the payload against the institutional schema, and writes only when the content hash differs from the last run.

Daily grant-portal poll control flow A daily cron trigger issues a GET to the grant portal endpoint. On a 429 or 5xx response the job applies exponential backoff with jitter and retries the GET; on a timeout or network error it logs the fault and exits non-zero; on a 200 it parses the JSON and validates it against the institutional schema. The validated payload hash is compared to the last saved state: if it matches, the run is an idempotent skip; if it has changed, the job performs an atomic CSV write with a state update and appends an audit-log entry. Daily cron trigger GET portal endpoint Parse JSON Validate schema Backoff + jitter Log fault, exit ≠ 0 Idempotent skip Atomic CSV write + update state file Append audit-log entry HTTP response Hash == last state? 429 / 5xx retry timeout / network 200 OK match changed
The daily poll enforces idempotency (hash check), fault tolerance (backoff with jitter), and schema validation before any write reaches disk.

Step 1 — Configure compliance-aligned logging and the frozen schema

Structured logging to a dedicated audit file is the foundation of non-repudiation. Pin the required-key set so schema drift is detected, not silently absorbed.

python
import os
import json
import logging
import hashlib
import csv
import tempfile
import shutil
from datetime import datetime, timezone
from typing import Dict, List
from pathlib import Path

import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

# Compliance-aligned structured logging: every run appends to an immutable
# audit file AND streams to stdout for the scheduler's job log.
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
    handlers=[
        logging.FileHandler("grant_poll_audit.log"),
        logging.StreamHandler(),
    ],
)
logger = logging.getLogger("grant_portal_poller")

# Frozen institutional schema — version-control this set with your policy config.
REQUIRED_SCHEMA_KEYS = {"grant_id", "pi_name", "award_amount", "status", "last_updated"}

Step 2 — Build a session with bounded retries and least-privilege headers

A single reusable requests.Session isolates authentication state and mounts an HTTPAdapter that retries only the transient status codes (429, 5xx) and only the idempotent GET method — never a write verb.

python
class GrantPortalPoller:
    def __init__(
        self,
        base_url: str,
        api_token: str,
        poll_interval_hours: int = 24,
        state_dir: str = ".compliance_state",
    ):
        self.base_url = base_url.rstrip("/")
        self.state_dir = Path(state_dir)
        self.state_dir.mkdir(parents=True, exist_ok=True)
        self.poll_interval = poll_interval_hours

        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {api_token}",   # token comes from env, never source
            "Accept": "application/json",
            "User-Agent": "UniversityResearchHub/1.0 (Compliance-Polling)",
        })

        # Exponential backoff for transient portal failures (429, 5xx).
        # allowed_methods is restricted to GET so a retry can never re-POST.
        retry_strategy = Retry(
            total=4,
            backoff_factor=1.5,
            status_forcelist=[429, 500, 502, 503, 504],
            allowed_methods=["GET"],
        )
        self.session.mount("https://", HTTPAdapter(max_retries=retry_strategy))

Step 3 — Fingerprint and validate the payload

The SHA-256 fingerprint is computed over a normalized (sorted) representation so cosmetic reordering by the portal never produces a false “changed” signal. Schema validation rejects malformed records before they can reach disk; rejected rows are logged, not committed.

python
class GrantPortalPoller:  # continued
    def _compute_payload_hash(self, data: List[Dict]) -> str:
        """Deterministic SHA-256 fingerprint for idempotency checks."""
        normalized = json.dumps(sorted(data, key=lambda x: x.get("grant_id")), sort_keys=True)
        return hashlib.sha256(normalized.encode("utf-8")).hexdigest()

    def _validate_schema(self, records: List[Dict]) -> List[Dict]:
        """Strict schema validation per institutional compliance standards."""
        valid = []
        for idx, record in enumerate(records):
            missing = REQUIRED_SCHEMA_KEYS - record.keys()
            if missing:
                logger.warning(f"Schema violation at index {idx}: missing keys {missing}")
                continue  # quarantine by omission — never write a partial record
            valid.append(record)
        return valid

Step 4 — Write atomically so a crash never leaves a half file

Writing to a temp file, fsync-ing, then shutil.move guarantees the output CSV is either the old version or the complete new one — never a truncated artifact an auditor would flag.

python
class GrantPortalPoller:  # continued
    def _atomic_write_csv(self, filepath: Path, records: List[Dict]) -> None:
        """Prevents partial writes and ensures audit-safe file operations."""
        with tempfile.NamedTemporaryFile(
            mode="w", delete=False, suffix=".csv", dir=filepath.parent
        ) as tmp:
            writer = csv.DictWriter(tmp, fieldnames=sorted(REQUIRED_SCHEMA_KEYS))
            writer.writeheader()
            writer.writerows(records)
            tmp.flush()
            os.fsync(tmp.fileno())          # force bytes to disk before rename
            shutil.move(tmp.name, str(filepath))  # atomic on the same filesystem

Step 5 — Orchestrate the idempotent daily poll

The driver ties the pieces together: fetch, validate, compare the fingerprint against the saved state, and write only on a genuine change. Network and decode failures return False so the scheduler sees a non-zero exit and can alert — but they never raise a false “data changed” event.

python
class GrantPortalPoller:  # continued
    def poll_and_sync(self, endpoint: str, output_csv: str) -> bool:
        """Execute idempotent daily poll with compliance logging."""
        url = f"{self.base_url}/{endpoint.lstrip('/')}"
        state_file = self.state_dir / f"{endpoint.replace('/', '_')}_state.json"
        output_path = Path(output_csv)

        try:
            response = self.session.get(url, timeout=(10, 30))  # (connect, read)
            response.raise_for_status()
            payload = response.json()
            records = payload if isinstance(payload, list) else payload.get("grants", [])
        except requests.exceptions.RequestException as e:
            logger.error(f"Network/portal failure: {e}")
            return False
        except ValueError as e:
            logger.error(f"JSON decode failure: {e}")
            return False

        valid_records = self._validate_schema(records)
        if not valid_records:
            logger.warning("No valid records after schema validation. Skipping write.")
            return True

        current_hash = self._compute_payload_hash(valid_records)
        previous_state = {}
        if state_file.exists():
            previous_state = json.loads(state_file.read_text())

        if previous_state.get("last_hash") == current_hash:
            logger.info("Idempotent check passed: payload unchanged. Skipping write.")
            return True

        self._atomic_write_csv(output_path, valid_records)
        new_state = {
            "last_hash": current_hash,
            "record_count": len(valid_records),
            "synced_at": datetime.now(timezone.utc).isoformat(),
            "endpoint": endpoint,
        }
        state_file.write_text(json.dumps(new_state, indent=2))
        logger.info(f"Compliant sync complete: {len(valid_records)} records -> {output_csv}")
        return True


# Execution entry point (schedule via cron/systemd)
if __name__ == "__main__":
    BASE_URL = os.getenv("GRANT_PORTAL_BASE_URL", "https://api.university-funding.example/v1")
    API_TOKEN = os.getenv("GRANT_PORTAL_API_TOKEN")
    if not API_TOKEN:
        raise RuntimeError("GRANT_PORTAL_API_TOKEN environment variable required")

    poller = GrantPortalPoller(base_url=BASE_URL, api_token=API_TOKEN)
    success = poller.poll_and_sync(endpoint="grants/active", output_csv="daily_grant_status.csv")
    exit(0 if success else 1)

Step 6 — Schedule the job

Run the poller once daily at 06:00, capturing both stdout and stderr into a rotating job log. A crontab entry:

bash
# m h  dom mon dow   command
0 6 * * *  GRANT_PORTAL_API_TOKEN=$(cat /etc/grant/token) /opt/grant/.venv/bin/python /opt/grant/poller.py >> /var/log/grant/poll.log 2>&1

Because each run re-checks the saved fingerprint, a missed day followed by a catch-up run is safe: the first successful poll reconciles state with no duplicate output.

Schema and field reference

The poller enforces this minimal required-key set. Type and constraint expectations align with sponsor data dictionaries; widen the set in your version-controlled policy config rather than in code.

Field Type Constraint Source rule
grant_id string Non-empty, unique per record; used as the sort key for hashing NIH/NSF award identifier (sponsor data dictionary)
pi_name string Non-empty NIH eRA Commons / NSF Research.gov PI of record
award_amount number ≥ 0, currency in USD 2 CFR 200 cost-principle reporting
status string Enumerated (e.g. active, pending, closed) Sponsor award lifecycle state
last_updated string ISO-8601 timestamp, normalized to UTC Audit-trail timestamp for non-repudiation

Verification

Confirm a run behaved correctly before trusting its output:

  1. Audit log: grant_poll_audit.log contains an INFO line — either Compliant sync complete: N records or Idempotent check passed — timestamped inside the scheduled window.
  2. Reproduce the hash: read the current CSV, re-run _compute_payload_hash on the parsed rows, and confirm it matches last_hash in .compliance_state/grants_active_state.json. An equal hash proves the on-disk data matches the recorded fingerprint.
  3. Row-count cross-check: the record_count in the state file must equal the data-row count of daily_grant_status.csv (minus the header) and the count shown in the portal UI snapshot.
  4. Dry-run idempotency: run the script twice back-to-back. The second run must log Idempotent check passed and leave the CSV’s mtime unchanged.

Troubleshooting

Three failure modes specific to daily portal polling:

  • Hash changes every run with no real data change. The portal is returning records in a non-deterministic order or with drifting last_updated timestamps. The sort in _compute_payload_hash handles reordering, but if timestamps churn, normalize or exclude volatile fields from the fingerprint before hashing.
  • Persistent 429 Too Many Requests despite backoff. Four retries with backoff_factor=1.5 may be too aggressive for a tightly rate-limited portal. Raise backoff_factor to 2.0, stagger the cron minute away from the top of the hour, or request a quota increase through your IT liaison. For high-volume reconciliation, move the workload onto the Async Processing & Queue Management path instead of a single synchronous poll.
  • Repeated Schema violation warnings. The portal changed its payload shape (API version drift). The poller correctly quarantines the rows by omission. Diff the live payload against your frozen REQUIRED_SCHEMA_KEYS, and if the primary endpoint is degraded, divert acquisition through your Fallback Routing Protocols until the schema is reconciled.

Frequently asked questions

Why hash the payload instead of trusting the portal's last_updated field?
Sponsor portals frequently bump last_updated on read or on internal re-indexing without any change to the substantive award data. A SHA-256 fingerprint over the normalized record set is the only reliable signal that the data an auditor cares about actually changed, which is what keeps the daily write idempotent.
Is it safe if the cron job runs twice in the same day?
Yes. The second run computes the same fingerprint, matches it against the saved state file, logs Idempotent check passed, and skips the write entirely. No duplicate CSV is produced and the audit log records that the data was unchanged.
Should the poller correct or interpret compliance status?
No. The polling layer only acquires, fingerprints, and validates the shape of the data. Interpretation and any state correction belong to human compliance review — the pipeline surfaces anomalies (schema violations, network faults) rather than resolving them silently.