Schema Validation Pipelines

Schema validation is the boundary control that decides which fare records are allowed to influence money. Within the broader Fare Data Ingestion & GTFS-RT Sync pipeline, it sits directly after raw acquisition and directly before reconciliation: every tap event, service alert, and trip update must pass a deterministic data contract before it can move revenue between routes, operators, or fare products. Treated as an afterthought, validation becomes a nightly cleanup job that silently discards records and distorts month-end close. Treated as a continuous control plane, it becomes the audit-defensible gate where malformed, duplicate, or out-of-order records are caught, quarantined, and accounted for the moment they arrive.

This component is owned jointly. The Python developer builds and instruments the validators, the transit operations engineer keeps them running through vendor firmware quirks and clock drift, and the revenue analyst depends on their completeness when reconciling farebox recovery ratios. Validation logic must also adapt to the velocity of each source: the low-latency stream from GTFS-RT Realtime Sync can only afford lightweight structural checks inline during protobuf deserialization — flagging missing trip_id references or an out-of-range position.latitude — while the batch feeds delivered by AFC API Data Extraction and CSV Batch Parsing Workflows can carry heavier semantic checks: verifying fare product codes against agency tariff tables, confirming tap-in/tap-out ordering, and rejecting duplicate transaction IDs before they reach the ledger.

The diagram below traces that control plane end to end: three acquisition sources feed a single memory-bounded validator that forks every record into a settlement path or a quarantine path, while metrics are tapped off continuously.

Prerequisites & environment

The patterns below assume a modern async Python stack and a canonical event shape agreed with the smart card schema mapping layer upstream. Raw hardware payloads should already be decoded to dictionaries before they reach these validators; this component enforces the contract, it does not parse card ciphers.

Assumptions this component makes explicit:

• Canonical timestamps are UTC. Validators normalize to UTC; they never assume the producer already did.
• Monetary fields are integer minor units (cents), never floating point. Float cents silently corrupt reconciliation totals under repeated addition.
• Every record carries a stable transaction_id. Idempotency and dead-letter replay both depend on it.
• AFC vendor exports vary. Expect BOM-prefixed UTF-8, inconsistent null tokens ("", "NULL", "\\N"), and occasional column reordering between firmware versions.

Memory-bounded streaming validation

High-volume tap streams routinely exceed what a synchronous, load-everything validator can hold. Materializing an entire JSON array or database dump into memory before validation invites queue saturation and OOM kills during peak ridership. The remedy is to parse and validate row-by-row using generators or async iterators, yielding validated batches while keeping only a bounded buffer resident.

Coupling a dialect-aware reader with incremental schema checks lets non-conforming records be quarantined immediately instead of accumulating. When the failure rate inside a sliding window crosses a threshold — the ratio below — the pipeline trips a circuit breaker rather than continuing to ingest corrupted data:

$$\text{failure\_ratio} = \frac{\sum_{i=1}^{N} \mathbb{1}[\text{record}_i \text{ invalid}]}{N} \;>\; \theta$$

import csv
import logging
from collections import deque
from typing import AsyncIterator, Dict, Any, List, Callable
from pydantic import ValidationError
from dataclasses import dataclass, field

logger = logging.getLogger(__name__)


@dataclass
class CircuitBreaker:
    """Trips OPEN when the failure count in a sliding window exceeds a threshold."""
    failure_threshold: int = 50
    window_size: int = 1000
    _failures: deque = field(default_factory=lambda: deque(maxlen=1000))
    _open: bool = False

    def record(self, success: bool) -> None:
        self._failures.append(not success)
        if sum(self._failures) > self.failure_threshold:
            self._open = True
            logger.critical("Validation circuit breaker OPEN. Halting stream.")

    def is_open(self) -> bool:
        return self._open


async def validate_csv_stream(
    file_path: str,
    schema_validator: Callable[[Dict[str, Any]], Any],
    breaker: CircuitBreaker,
    batch_size: int = 500,
) -> AsyncIterator[List[Any]]:
    """Memory-bounded CSV validator that yields validated batches with circuit breaking.

    Reads with utf-8-sig to strip BOM-prefixed exports from legacy AFC back offices,
    and quarantines (rather than raises on) individual malformed rows.
    """
    # utf-8-sig transparently strips a leading BOM emitted by many vendor exports.
    with open(file_path, "r", encoding="utf-8-sig", newline="") as f:
        reader = csv.DictReader(f)
        batch: List[Any] = []

        for row in reader:
            if breaker.is_open():
                raise RuntimeError("Circuit breaker tripped. Stream aborted.")

            try:
                validated = schema_validator(row)
                batch.append(validated)
                breaker.record(True)
            except ValidationError as exc:
                breaker.record(False)
                logger.warning("Schema violation quarantined: %s", exc)
                continue
            except (ValueError, KeyError) as exc:
                breaker.record(False)
                logger.error("Malformed row rejected: %s", exc)
                continue

            if len(batch) >= batch_size:
                yield batch
                batch = []

        if batch:
            yield batch

The generator never holds more than batch_size validated records plus the breaker’s fixed-length window, so memory stays flat regardless of file size. Failures are catalogued, not swallowed: each rejected row increments the breaker and emits a structured log line the audit trail can replay.

Schema validation & transit-specific edge cases

The most robust approach for transit data engineers centers on declarative schema definitions that map directly to AFC event structures. Type-driven contracts enforce strict rules without hand-written per-field checks, and Pydantic v2’s @field_validator and @model_validator decorators let engineers encode transit’s awkward realities directly into the model. The full model catalogue lives in Implementing Pydantic Models for AFC Event Streams; the core shape is below.

from datetime import datetime, timezone
from typing import Optional
from enum import Enum
from pydantic import BaseModel, Field, field_validator, model_validator


class FareMediaType(str, Enum):
    CONTACTLESS = "contactless"
    SMART_CARD = "smart_card"
    MOBILE_QR = "mobile_qr"
    PAPER_TICKET = "paper_ticket"


class TapEvent(BaseModel):
    transaction_id: str = Field(..., min_length=16, max_length=36)
    media_type: FareMediaType
    route_id: str
    stop_id: str
    tap_timestamp: datetime
    fare_amount_cents: int = Field(..., ge=0)  # integer minor units, never float
    direction: int = Field(..., ge=0, le=1)
    previous_tap_id: Optional[str] = None

    @field_validator("tap_timestamp")
    @classmethod
    def enforce_utc(cls, v: datetime) -> datetime:
        # Reject naive datetimes outright; a missing tz is a producer bug, not a default.
        if v.tzinfo is None:
            raise ValueError("tap_timestamp must be timezone-aware; got naive datetime")
        return v.astimezone(timezone.utc)

    @field_validator("fare_amount_cents")
    @classmethod
    def validate_tariff_bounds(cls, v: int) -> int:
        MAX_CENTS = 2500  # agency-specific single-trip ceiling ($25.00)
        if v > MAX_CENTS:
            raise ValueError(f"Fare {v} cents exceeds tariff ceiling {MAX_CENTS}")
        return v

    @model_validator(mode="after")
    def check_epoch_sanity(self) -> "TapEvent":
        if self.tap_timestamp.timestamp() <= 0:
            raise ValueError("Non-positive epoch timestamp indicates clock failure")
        return self

Transit data breaks naive validators in predictable ways. The contract has to absorb each of these without dropping data on the floor:

• Null handling. Vendor CSVs encode “missing” inconsistently. Normalize "", "NULL", and "\N" to None in a pre-validator so an empty stop_id fails a real rule rather than passing as a literal string.
• Encoding fallback. Reading with utf-8-sig strips the BOM many back offices prepend; wrap the open in a fallback to latin-1 for archival files that predate a UTF-8 mandate, and log the fallback so it is auditable.
• Timezone normalization. Offline validators stamp local wall-clock time; enforce_utc refuses naive datetimes so daylight-saving transitions can’t create phantom “future” or duplicated taps.
• Idempotency. Feed retries and at-least-once queues redeliver the same transaction_id. Deduplication has to happen inside validation, not downstream, or the same tap settles twice.
• Out-of-order taps. Devices that were offline flush their buffer late. A tap-out can arrive before its tap-in is visible; the reconciliation state machine below treats an unmatched tap-out as an orphan rather than an error.

Error handling & reconciliation logic

Validation failures must never block the pipeline. Instead they produce structured error payloads that feed an exception queue and audit trail. A production-grade pipeline routes valid events into a reconciliation buffer while sending malformed records to a dead-letter queue (DLQ) with full context preserved for replay and forensic review.

Reconciliation itself is deterministic aggregation: matching tap-in/tap-out pairs, applying the transfer window logic agreed per operator, and attributing net fare. Partial journeys are handled gracefully — orphaned tap-ins age out after a configurable window (typically 120 minutes) and are flagged for manual audit rather than silently discarded.

The state machine below traces a fare session from tap-in through settlement, expiry, or orphan:

Validation failures and orphaned taps follow the dual-path routing below, separating settled revenue from records that need audit:

import hashlib
import logging
from typing import List, Dict

logger = logging.getLogger(__name__)


class ReconciliationEngine:
    def __init__(self, transfer_window_minutes: int = 120) -> None:
        self.transfer_window = transfer_window_minutes * 60
        self.active_sessions: Dict[str, Dict] = {}
        self.settled_trips: List[Dict] = []
        self.dlq: List[Dict] = []
        self._seen_tx_ids: set = set()

    def process_batch(self, validated_events: List[TapEvent]) -> None:
        for event in validated_events:
            try:
                self._route_event(event)
            except (KeyError, ValueError) as exc:
                self._to_dlq(event, str(exc))

    def _to_dlq(self, event: TapEvent, reason: str) -> None:
        self.dlq.append({
            "event": event.model_dump(mode="json"),
            "error": reason,
            "hash": hashlib.sha256(event.transaction_id.encode()).hexdigest(),
        })

    def _route_event(self, event: TapEvent) -> None:
        # Idempotent dedup via an O(1) seen-set (safe under at-least-once feed retries).
        if event.transaction_id in self._seen_tx_ids:
            return
        self._seen_tx_ids.add(event.transaction_id)

        # A tap-in opens a session keyed by its own transaction_id; the matching
        # tap-out references it via previous_tap_id, so the lookup key differs by
        # tap direction.
        if event.previous_tap_id is None:
            self._open_session(event)
            return

        session_key = f"{event.media_type}:{event.previous_tap_id}"
        session = self.active_sessions.get(session_key)
        if session is None:
            # Orphaned tap-out (out-of-order flush or lost tap-in).
            self._to_dlq(event, "Orphaned tap-out without matching session")
            return

        delta = event.tap_timestamp.timestamp() - session["created_at"]
        if delta <= self.transfer_window:
            self.settled_trips.append({
                "in": session["in"],
                "out": event,
                "duration_sec": delta,
                "net_fare_cents": session["in"].fare_amount_cents,
            })
            del self.active_sessions[session_key]
        else:
            # Expired: treat the late tap-out as a fresh journey start.
            del self.active_sessions[session_key]
            self._open_session(event)

    def _open_session(self, event: TapEvent) -> None:
        self.active_sessions[f"{event.media_type}:{event.transaction_id}"] = {
            "in": event,
            "out": None,
            "created_at": event.tap_timestamp.timestamp(),
        }

    def flush_expired_sessions(self, current_utc_ts: float) -> List[TapEvent]:
        """Age out incomplete journeys for audit."""
        expired: List[TapEvent] = []
        for key, session in list(self.active_sessions.items()):
            if current_utc_ts - session["created_at"] > self.transfer_window:
                expired.append(session["in"])
                del self.active_sessions[key]
        return expired

Integrating with adjacent pipeline stages

Schema validation is a stage, not a silo. It receives already-decoded records from the acquisition components and emits two typed streams — canonical events and dead-letter entries — that downstream stages consume.

• Inbound from acquisition. AFC API Data Extraction and CSV Batch Parsing Workflows hand off raw dictionaries; validation is the first stage that can safely assume a typed TapEvent. Keep the contract in one module so both feeds import the same models.
• Sideband to real-time sync. Settled trips carry a route_id and stop_id that GTFS-RT Realtime Sync uses to align fare events with the vehicle trajectory that produced them — validation guarantees those keys are present and well-formed before the spatial-temporal match runs.
• Downstream to rule engines. Clean events feed the fare rule validation & calculation engines, where capping and concession rules are applied. Records that failed validation while a device ran offline should be reconciled against the agency’s fallback routing strategies so locally cached taps are not double-counted on reconnect.

The handoff contract is deliberately narrow: adjacent stages depend on the shape (TapEvent) and the guarantees (UTC, deduplicated, tariff-bounded), never on the internal breaker or buffer state.

Performance & scale considerations

At three million taps a day a naive validator is the pipeline’s bottleneck long before the database is. The levers that matter for fare-data volumes:

• Chunk sizing. A batch_size of 500–1000 balances per-batch overhead against buffer residency. Larger batches amortize downstream writes but raise the memory floor and the blast radius of a poison record.
• Memory bounds. The streaming generator plus a fixed-length breaker window keeps resident memory independent of file size — the single most important property for multi-gigabyte overnight exports.
• Parallelism caveats. Validation parallelizes cleanly across shards only if deduplication is partitioned by transaction_id (or media_hash) so two workers never race on the same tap. The reconciliation active_sessions map, by contrast, must be partitioned by media/card so a rider’s tap-in and tap-out land on the same worker — otherwise pairs never match. Shard on the card key, not round-robin.
• Pydantic v2 speed. The Rust-backed core validates roughly an order of magnitude faster than v1; prefer model_validate over constructing dicts by hand, and avoid re-compiling models inside the hot loop.

Operational checklist

Production readiness for a transit-ops deployment, in order:

1. Expose validation_success_rate, queue_depth, reconciliation_latency, and orphaned_session_count as Prometheus gauges or OpenTelemetry traces.
2. Alert on circuit-breaker state transitions and on DLQ volume crossing a per-feed threshold.
3. Wire a drain mode that pauses new ingestion when the DLQ backlog spikes, so corrupted records can’t cascade into settlement.
4. Pin Pydantic and Python versions in the lockfile; a minor validation-semantics change silently reshapes what reaches the ledger.
5. Persist every DLQ entry with full payload and error context for replay, and test the replay path — a DLQ you cannot drain is a data-loss incident in slow motion.
6. Reconcile validated + quarantined = ingested every run; a gap means records vanished between stages.
7. Validate models against the official GTFS Realtime Specification and the current Pydantic documentation on each dependency bump.

By treating schema validation as a continuous, memory-bounded control plane rather than a batch cleanup step, agencies achieve deterministic revenue attribution, reduce audit overhead, and keep the pipeline resilient under peak ridership.

Deeper implementation guides

• Implementing Pydantic Models for AFC Event Streams — the full model catalogue behind the TapEvent contract above.

Related

• AFC API Data Extraction — the upstream stage that feeds raw records into validation.
• CSV Batch Parsing Workflows — dialect-aware batch reading that pairs with the streaming validator.
• GTFS-RT Realtime Sync — consumes validated events for spatial-temporal matching.
• Smart Card Schema Mapping — decodes raw card payloads into the dictionaries this stage validates.
• Transfer Window Logic — the per-operator rules the reconciliation engine applies.

Part of Fare Data Ingestion & GTFS-RT Sync.