AFC API Data Extraction

Automated Fare Collection (AFC) systems generate the financial and operational backbone of modern transit networks. Extracting this data reliably requires more than simple HTTP polling; it demands a disciplined pipeline architecture that aligns fare telemetry with service performance. Within the broader Fare Data Ingestion & GTFS-RT Sync framework, API extraction is the critical first hop — the point where vendor-specific endpoints become structured, auditable records ready for revenue reconciliation and ridership analytics. Get this hop wrong and every downstream number, from farebox recovery ratios to inter-agency settlements, inherits the error.

This is the extraction component that pulls tap-level events out of a vendor’s back-office API, applies backpressure so it survives peak-hour surges, validates each record at the boundary, and hands clean events off to the reconciliation ledger. It is owned jointly: the Python automation developer builds and instruments it, the transit operations engineer keeps it running through vendor throttling and outages, and the revenue analyst depends on its completeness at month-end close.

Prerequisites & Environment

This component targets Python 3.11+ (for asyncio.TaskGroup, tomllib, and faster exception groups). The reference implementation depends on a small, production-proven stack:

Vendor assumptions. AFC back-office APIs vary widely. This component assumes an HTTP endpoint that returns transaction pages via limit/offset or cursor pagination, authenticates with a bearer token or HMAC signature, and enforces some rate ceiling. Truly legacy SOAP or fixed-window gateways need the pacing and checkpointing patterns detailed in Handling Rate Limits on Legacy AFC Vendor APIs; flat-file exporters are handled instead by the CSV Batch Parsing Workflows component.

Schema expectations. Every tap payload is expected to carry, at minimum, a transaction identifier, a device identifier, a UTC-convertible timestamp, a fare_media_type, an origin stop reference, and a monetary amount. The meaning of fare_media_type and its media-specific fields comes from Smart Card Schema Mapping — extraction binds raw hardware fields to that canonical taxonomy rather than inventing its own.

Memory-Efficient Async Extraction

Transit operators and mobility developers must navigate heterogeneous vendor APIs, each with distinct authentication schemes, pagination models, and rate ceilings. When tap volumes spike during peak hours or special events, synchronous extraction quickly becomes a bottleneck. The production standard is async batching with strict backpressure controls. Buffering raw JSON payloads in-memory without streaming to disk or a message broker will inevitably crash ingestion workers during end-of-day reconciliation runs.

Implement a producer-consumer architecture using bounded asyncio.Queue objects and chunked HTTP responses. Stream payloads directly to NDJSON files or a lightweight broker to maintain a constant memory footprint regardless of transaction volume.

The sequence below shows the producer-consumer flow that keeps memory bounded under peak load:

import asyncio
import aiohttp
from typing import AsyncGenerator

async def stream_tap_batches(
    session: aiohttp.ClientSession,
    base_url: str,
    params: dict,
    batch_size: int = 500,
) -> AsyncGenerator[dict, None]:
    """Fetch paginated AFC data and yield records without full in-memory buffering."""
    offset = 0
    while True:
        async with session.get(
            base_url, params={**params, "limit": batch_size, "offset": offset}
        ) as resp:
            resp.raise_for_status()
            # Buffer only one bounded page at a time, then yield records individually
            payload = await resp.json()
            if not payload.get("transactions"):
                break

            for tx in payload["transactions"]:
                yield tx

            offset += batch_size
            # Yield control to the event loop to prevent starvation
            await asyncio.sleep(0)

The generator above is the producer. In a full deployment it feeds a bounded asyncio.Queue, and one or more consumer tasks drain that queue to an NDJSON file or broker topic. Because queue.put() awaits when the queue is full, a slow sink automatically throttles the fetcher — the system self-regulates instead of accumulating unbounded pages in RAM. Pair this with Python’s asyncio documentation best practices for task grouping, graceful cancellation, and semaphore-limited concurrency to prevent socket exhaustion.

Rate Limiting & Resilience Patterns

Legacy back-office systems often expose REST or SOAP endpoints that lack native streaming capabilities, forcing engineers to implement request pacing through exponential backoff, token rotation, and request windowing. Transit APIs frequently return 429 Too Many Requests or 503 Service Unavailable during farebox sync windows, and the exact throttling behavior — token bucket, fixed window, or silent TCP reset — is covered vendor-by-vendor in Handling Rate Limits on Legacy AFC Vendor APIs.

Production extractors must implement jittered backoff, circuit breakers, and idempotent request signatures. The tenacity library provides a clean abstraction for retry policies, but you must explicitly handle vendor-specific rate-limit headers (X-RateLimit-Remaining, Retry-After).

import logging
import httpx
from tenacity import (
    retry,
    stop_after_attempt,
    wait_exponential,
    retry_if_exception_type,
)

logger = logging.getLogger("afc.extract")

@retry(
    retry=retry_if_exception_type((httpx.HTTPStatusError, httpx.ConnectError)),
    wait=wait_exponential(multiplier=1, min=2, max=30),
    stop=stop_after_attempt(5),
    reraise=True,
)
async def resilient_fetch(
    client: httpx.AsyncClient, endpoint: str, headers: dict
) -> dict:
    """Fetch one AFC page with bounded retries; surface Retry-After for observability."""
    resp = await client.get(endpoint, headers=headers)
    if resp.status_code == 429:
        # Surface the server's Retry-After hint for observability; tenacity applies
        # exponential backoff between attempts via the wait policy above.
        retry_after = resp.headers.get("Retry-After", "2")
        logger.warning("Rate limited on %s; Retry-After=%ss", endpoint, retry_after)
        raise httpx.HTTPStatusError(
            f"Rate limited. Retry-After: {retry_after}s",
            request=resp.request,
            response=resp,
        )
    resp.raise_for_status()
    return resp.json()

Because retried requests may re-send the same page, the fetch must be idempotent: derive a stable request signature (endpoint plus cursor/offset) and let the downstream dedup gate reject any records that arrive twice. Never let a retry mutate the checkpoint cursor before the page is confirmed persisted.

Schema Validation & Transit-Specific Edge Cases

Raw fare transactions are meaningless without spatial and temporal context, but they are also dangerous without strict typing. Every extracted payload must pass through a validation gate that enforces data types, required fields (fare_media_type, tap_timestamp, origin_stop_id), and business rules (no negative balances, sequential tap constraints, valid media types). Pydantic v2 is the standard for this layer thanks to its compiled validation speed and explicit error serialization; the reusable event models themselves live in Schema Validation Pipelines, and the concrete AFC model definitions are walked through in Implementing Pydantic Models for AFC Event Streams.

Invalid records must be quarantined for manual review rather than poisoning downstream revenue models. Implement a dual-sink routing pattern that separates clean data from malformed payloads.

The dual-sink validation gate below routes each payload to exactly one destination:

Monetary amounts are the one field you must never model as a binary float: 0.10 + 0.20 does not equal 0.30 in IEEE-754, and that drift compounds across millions of taps into settlement disputes. Use decimal.Decimal end to end — Pydantic v2 validates and serializes it natively.

import logging
from datetime import datetime, timezone
from decimal import Decimal
from typing import Protocol

from pydantic import BaseModel, ValidationError, field_validator

logger = logging.getLogger("afc.validate")

class TapSink(Protocol):
    async def __call__(self, record: dict) -> None: ...

class TapRecord(BaseModel):
    transaction_id: str
    tap_timestamp: datetime
    device_id: str
    origin_stop_id: str
    fare_media_type: str
    amount: Decimal  # money is Decimal, never float

    @field_validator("tap_timestamp")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        # Naive timestamps are assumed vendor-local only if a tz is configured
        # elsewhere; here we require awareness and normalize to UTC.
        if v.tzinfo is None:
            raise ValueError("tap_timestamp must be timezone-aware")
        return v.astimezone(timezone.utc)

    @field_validator("amount")
    @classmethod
    def enforce_non_negative(cls, v: Decimal) -> Decimal:
        if v < Decimal("0"):
            raise ValueError("Negative fare amount violates revenue integrity rules")
        return v

async def validate_and_route(
    raw: dict,
    valid_sink: TapSink,
    quarantine_sink: TapSink,
) -> None:
    """Validate one raw payload and route it to exactly one sink."""
    try:
        parsed = TapRecord.model_validate(raw)
    except ValidationError as exc:
        tx_id = raw.get("transaction_id", "UNKNOWN")
        logger.warning("Quarantined tx %s: %s", tx_id, exc)
        await quarantine_sink({**raw, "validation_error": str(exc)})
        return
    await valid_sink(parsed.model_dump(mode="json"))

Beyond typing, four transit-specific edge cases decide whether the extractor is trustworthy:

1. Timezone normalization. Convert every tap_timestamp to UTC at the boundary, as the validator above does. DST transitions are the classic cause of duplicated or missing revenue attribution — a naive local timestamp during the autumn fall-back hour maps to two real instants.
2. Null and missing fields. Distinguish a genuinely absent field from a vendor’s sentinel ("", "0", "N/A"). Coerce known sentinels to None before validation so the quarantine reason is meaningful rather than a generic type error.
3. Encoding fallback. Some fareboxes emit Latin-1 or Windows-1252 stop names inside a nominally UTF-8 payload. Decode defensively (utf-8 then latin-1 fallback) rather than letting a single mojibake byte abort a whole page.
4. Idempotency. Assign each record a deterministic dedup key (device_id + tap_timestamp_utc + sequence) so replays from a retried page or a resumed checkpoint are rejected downstream instead of double-counted.

Refer to the Pydantic v2 documentation for advanced validators, custom error formatting, and integration with message brokers like Kafka or Redis Streams.

Integration: GTFS Alignment & Reconciliation Handoff

Extracted tap records are only half of an attributable event; they must be cross-referenced against scheduled and real-time vehicle positions. The GTFS-RT Realtime Sync component provides the trip_id, vehicle_id, and timestamp anchors needed to map fare events to actual service delivery. AFC APIs rarely return GTFS-aligned identifiers natively, so extraction must emit a stable device_id and precise UTC timestamp, and let the sync stage resolve them against a position cache. This is the clean seam between components: extraction owns capture and validation; sync owns spatial-temporal matching.

Handoff at scale relies on three deterministic rules:

1. Timezone-anchored keys. Because taps are already UTC-normalized here, the sync stage can window against GTFS-RT timestamps without re-deriving offsets.
2. Spatial proximity matching. When GPS drift occurs, taps are matched to the nearest valid stop_id within a configurable radius (for example 50 m) using Haversine distance or PostGIS ST_DWithin — the same zone geometry described in Mapping Multi-Modal Fare Zones to PostGIS Polygons.
3. Temporal windowing. Taps join to GTFS-RT vehicle positions using a sliding window (tap_timestamp ± 90s) to absorb farebox sync latency and cellular delay.

When the real-time feed degrades or a validator is offline, taps should not be dropped — they follow the Fallback Routing Strategies pattern, caching locally and replaying against the GTFS-static schedule once connectivity returns. For authoritative field semantics, consult the official GTFS Realtime Specification.

Performance & Scale Considerations

A mid-size agency processing three million taps per day pushes real volume through this component, and the failure modes are all about bounded resources rather than raw throughput:

• Chunk sizing. Page size trades request overhead against per-page memory. Start at 500 records; raise it only until a single page comfortably fits the consumer’s per-record processing budget. Oversized pages defeat backpressure because one await resp.json() materializes the whole chunk.
• Memory bounds. Keep the asyncio.Queue maxsize small (single-digit pages). The invariant is: at most maxsize + workers pages resident at once, independent of total daily volume.
• Parallelism caveats. Bound concurrency with a semaphore sized to the vendor’s rate ceiling, not your core count. Fan-out beyond the ceiling only converts throughput into 429s and wasted backoff. Connection pools should be capped for the same reason — an unbounded aiohttp connector will exhaust sockets before it exhausts the API.
• Bulk batch sources. Not every vendor offers a live API; many deliver multi-gigabyte CSV dumps over SFTP. Never call pandas.read_csv() on a whole daily file — use chunked iterators or polars.scan_csv() for O(1) memory, as benchmarked in Optimizing pandas chunksize for 10M-Row Fare Files. Always reconcile row counts against the vendor manifest before committing.
• Security under load. Retained raw request/response dumps can leak card identifiers; mask media fields before persisting and follow the retention and encryption rules in AFC System Security Boundaries so scaling the pipeline never scales a compliance breach.

Operational Deployment Checklist

Work through these before promoting the extractor to a production transit-ops deployment:

1. Implement bounded async queues and NDJSON streaming to prevent out-of-memory failures during peak tap surges.
2. Configure exponential backoff with jitter and explicit Retry-After header parsing; verify idempotent replay on retried pages.
3. Enforce Pydantic schema validation at the ingestion boundary and route every failure to a dedicated quarantine sink with the original payload and error.
4. Model all monetary amounts as Decimal, and normalize every tap_timestamp to UTC before applying business rules.
5. Maintain a versioned device_id → stop_id mapping table with an audit trail for hardware swaps.
6. Run a daily reconciliation diff of SUM(validated_amounts) against the vendor settlement report, and alert on variance.
7. Instrument extraction latency, 429/5xx rates, and quarantine volume with Prometheus/Grafana dashboards.
8. Rotate API credentials on a schedule and mask card-identifying fields in any retained request logs.

Extend This Component

The most common next build on top of extraction is incremental sync: replacing full offset-based pulls with cursor-based pagination so each run only fetches taps newer than the last checkpoint. The vendor-throttling foundations for that live in Handling Rate Limits on Legacy AFC Vendor APIs, which pairs stateful checkpointing with defensive parsing for exactly-once semantics under aggressive rate caps.

Related

• Handling Rate Limits on Legacy AFC Vendor APIs→
• Schema Validation Pipelines→
• GTFS-RT Realtime Sync→
• CSV Batch Parsing Workflows→
• Smart Card Schema Mapping→

↑ Part of Fare Data Ingestion & GTFS-RT Sync