Schema Validation with Pydantic for Traffic Data

A parser can only guarantee that a row is structurally intact — the right columns, decoded cleanly, timezone-aware. It cannot tell you that a :30 makegood is missing its reference spot, that an overnight commercial was booked into a morning-drive daypart, or that a billing code no longer exists in the rate card. When those semantic defects slip past ingestion, they do not fail loudly; they surface hours later as a missed airing, a double-booked break, or a reconciliation gap that a station has to credit back. This guide owns the semantic gate of the Avion & Avstar Ingestion Pipelines architecture: the layer where a structurally valid record is turned into a business-valid, scheduler-ready spot — or is rejected with a precise, auditable reason. It is written for the Python engineers who build the validation service, and for the broadcast traffic managers who have to trust that everything past this gate is safe to air.

Concept & Data Model #

Validation sits immediately downstream of extraction. The parsing Avion export formats layer resolves delimiters, normalizes encoding, and converts wall-clock times to UTC, then hands off a stream of typed-but-untrusted records. This layer applies every business rule the parser deliberately deferred, so the two failure classes never mix: a truncated file is a parsing incident, while an invalid daypart is a validation incident, and each routes to a different remediation queue.

The unit of validation is a single spot, modelled against the canonical spot schema and metadata. The table below is the field-level contract the Pydantic model enforces at the boundary; every constraint here is a rule that, if violated, must stop the record before it reaches the scheduling queue.

Two properties make this model a gate rather than a type check. First, it must be strict: model_config = ConfigDict(strict=True) disables silent coercion, so a :30 string never quietly becomes the integer 30 unless a validator explicitly permits it. Traffic managers depend on deterministic behaviour — a spot length is either exactly thirty seconds or it is a rejection, never an ambiguous coercion. Second, cross-field rules run only after every field has parsed, so the model never emits a half-validated object.

Figure — Each structured record passes strict field validation then cross-field business rules; only records that clear both gates reach the scheduling queue, while every failure is quarantined with a structured, correlated error.

Implementation Approach #

Three design decisions dominate a production Pydantic validation service, and each is a trade-off worth making on purpose.

Field constraints belong in the type; business rules belong in validators. Anything expressible as a bound — length, range, pattern, enum membership — is declared with Field(...) so Pydantic’s Rust-backed core enforces it before any Python runs. Rules that span fields — a makegood needing its reference, an airtime hour needing to fall inside its declared daypart — belong in model_validator(mode="after"), which fires only once all fields exist. Keeping the two tiers separate means a field-level rejection is cheap and a cross-field rejection is unambiguous about which invariant broke.

Validate strictly, but normalize deliberately. Strict mode is the default posture, yet real Avion exports carry broadcast shorthand — :30 for thirty seconds, lower-case client codes. Rather than relaxing strict=True globally, a narrow mode="before" validator normalizes exactly those known dialects and nothing else. This preserves determinism: coercion happens only where a documented convention justifies it, and every other ambiguous value still fails loudly.

Treat a rejection as data, not an exception. A validation failure is a structured event with a field, a raw value, a violated constraint, and a correlation ID — not a stack trace to be logged and forgotten. Deterministic failures (invalid daypart, missing billing code) go to a dead-letter queue for a traffic manager to triage; only transient infrastructure faults get retried. This is the same boundary discipline the Avstar API authentication and rate limits layer applies upstream, and it is why high-volume drops flow through the chunked async batch processing for high-volume logs pattern instead of instantiating thousands of models at once. Billing codes carry an extra rule: the validator only checks that a code is present and well-formed, deferring canonicalization to the standardizing billing codes across traffic systems stage so that a single source of truth owns the mapping.

Production Python Implementation #

The module below is deployable rather than illustrative. It defines the strict spot model, normalizes broadcast shorthand, enforces makegood and daypart invariants, and — critically — never raises past its own boundary: it returns either a validated model or a structured ValidationOutcome that carries everything a quarantine queue needs. Every decision emits a log line in the traffic-ops format timestamp | level | module | spot_id.

from __future__ import annotations

import logging
from datetime import datetime, timezone
from enum import Enum
from typing import Iterable, Iterator, Optional, Union

from pydantic import (
    BaseModel,
    ConfigDict,
    Field,
    ValidationError,
    field_validator,
    model_validator,
)

# Traffic-ops structured log format: timestamp | level | module | spot_id
logging.basicConfig(
    format="%(asctime)s | %(levelname)s | %(name)s | %(spot_id)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
logger = logging.getLogger("avstar.validate")


def _log(level: int, msg: str, spot_id: str = "-") -> None:
    """Emit a log line carrying the spot_id — the pivot key for every traffic audit."""
    logger.log(level, msg, extra={"spot_id": spot_id})


class Daypart(str, Enum):
    MORNING_DRIVE = "MORN"
    MIDDAY = "MIDD"
    AFTERNOON_DRIVE = "AFTD"
    EVENING = "EVEN"
    OVERNIGHT = "OVN"


class SpotType(str, Enum):
    COMMERCIAL = "COM"
    PSA = "PSA"
    MAKEGOOD = "MG"
    BUMPER = "BUMP"


# Daypart windows are [start, end) in station-local broadcast hours.
_DAYPART_HOURS: dict[Daypart, tuple[int, int]] = {
    Daypart.MORNING_DRIVE: (6, 10),
    Daypart.MIDDAY: (10, 15),
    Daypart.AFTERNOON_DRIVE: (15, 19),
    Daypart.EVENING: (19, 24),
    Daypart.OVERNIGHT: (0, 6),
}


class TrafficSpot(BaseModel):
    """The semantic contract for one airing. Strict by design: no silent coercion."""

    model_config = ConfigDict(strict=True, populate_by_name=True, extra="forbid")

    spot_id: str = Field(..., min_length=6, max_length=24, pattern=r"^[A-Z0-9\-]+$")
    client_code: str = Field(..., min_length=2, max_length=8)
    campaign_id: str = Field(..., min_length=1)
    spot_type: SpotType
    daypart: Daypart
    scheduled_airtime: datetime
    duration_seconds: int = Field(..., ge=5, le=120)
    billing_code: str = Field(..., min_length=1)
    is_live_read: bool = False
    makegood_ref: Optional[str] = None
    priority_tier: int = Field(default=1, ge=1, le=5)

    @field_validator("duration_seconds", mode="before")
    @classmethod
    def normalize_duration(cls, v: Union[str, int]) -> int:
        # Accept broadcast-style ":30" shorthand as well as plain integers,
        # but reject anything that is not a clean whole-second value.
        if isinstance(v, str):
            token = v.strip().lstrip(":")
            if not token.isdigit():
                raise ValueError(f"non-integer duration token: {v!r}")
            return int(token)
        return v

    @field_validator("client_code", mode="before")
    @classmethod
    def uppercase_client(cls, v: str) -> str:
        # Client codes are case-insensitive in Avion but canonicalized upper-case.
        return v.strip().upper() if isinstance(v, str) else v

    @field_validator("scheduled_airtime")
    @classmethod
    def require_tz_aware(cls, v: datetime) -> datetime:
        # A naive airtime is unschedulable — the parser must have normalized to UTC.
        if v.tzinfo is None:
            raise ValueError("scheduled_airtime must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)

    @model_validator(mode="after")
    def validate_makegood_logic(self) -> "TrafficSpot":
        # A makegood must reference the spot it compensates; nothing else may.
        if self.spot_type is SpotType.MAKEGOOD and not self.makegood_ref:
            raise ValueError("MAKEGOOD spots require a makegood_ref")
        if self.spot_type is not SpotType.MAKEGOOD and self.makegood_ref:
            raise ValueError("makegood_ref is only valid on MAKEGOOD spots")
        return self

    @model_validator(mode="after")
    def enforce_daypart_alignment(self) -> "TrafficSpot":
        # Airtime hour must fall inside the declared daypart window.
        hour = self.scheduled_airtime.hour
        start, end = _DAYPART_HOURS[self.daypart]
        if not (start <= hour < end):
            raise ValueError(
                f"airtime hour {hour:02d} does not fall in daypart {self.daypart.value}"
            )
        return self


class ValidationOutcome(BaseModel):
    """Structured result the DLQ consumes when validation fails — never a stack trace."""

    spot_id: str
    ok: bool
    errors: list[dict[str, str]] = Field(default_factory=list)


def validate_record(raw: dict, correlation_id: str) -> Union[TrafficSpot, ValidationOutcome]:
    """Validate one structured record; return a model on success or a routed outcome."""
    spot_id = str(raw.get("spot_id", "-"))
    try:
        spot = TrafficSpot.model_validate(raw)
    except ValidationError as exc:
        errors = [
            {
                "field": ".".join(str(p) for p in e["loc"]),
                "value": repr(e.get("input")),
                "constraint": e["msg"],
                "correlation_id": correlation_id,
            }
            for e in exc.errors()
        ]
        _log(logging.ERROR, f"rejected: {len(errors)} violation(s)", spot_id)
        return ValidationOutcome(spot_id=spot_id, ok=False, errors=errors)
    _log(logging.INFO, f"validated {spot.spot_type.value} for daypart {spot.daypart.value}", spot.spot_id)
    return spot


def validate_batch(
    records: Iterable[dict], correlation_id: str
) -> Iterator[Union[TrafficSpot, ValidationOutcome]]:
    """Stream validation over a batch, yielding results without accumulating in memory."""
    for raw in records:
        yield validate_record(raw, correlation_id)

The validate_batch generator is what keeps the service memory-stable under load: it yields each outcome as it is produced instead of building a list, so a validated model can be streamed straight to the scheduling queue and a ValidationOutcome straight to the dead-letter queue, without the whole drop ever residing in the heap at once. Pydantic v2’s Rust core does the field-level work; the Python layer only expresses the broadcast rules and the routing.

Validation & Edge Cases #

The rules above are the common path. Production breaks on the boundary conditions that generic model code never anticipates, and each of these deserves an explicit test fixture.

• Daypart boundaries at the hour edge. The daypart windows are half-open [start, end), so a spot at exactly 10:00 belongs to Midday, not Morning Drive. The < end comparison is deliberate — an inclusive upper bound would let a single airtime satisfy two dayparts and make the rule non-deterministic.
• Overnight wrap-around. The Overnight window (0, 6) never crosses midnight in UTC because the parser already resolved local time; a spot that looks like it should be overnight but carries an 04:00 UTC hour after a timezone shift is a real alignment failure, not a rule bug, and must reject rather than be auto-corrected.
• Sports overruns and makegood chains. When a live event runs long, the original spot is preempted and a makegood is issued. The validator enforces that the makegood carries a makegood_ref, but it must not resolve whether that reference points at a real preempted spot — that lookup, and the routing that follows, is owned by make-good routing for preemptions. Validation proves the shape; scheduling proves the linkage.
• Zero and fractional durations. A :00 or an empty length is a structural rejection, and a 27.5-second value is refused outright — strict mode never truncates a float to an int. Frame-accurate second boundaries are the subject of the dedicated validating spot durations against broadcast standards walkthrough.
• Priority tiers and competitive separation. priority_tier is validated as an ordinal 1–5 but never renumbered here; the ordinal is what oversold-break preemption and competitive-separation windows key off downstream, so preserving it verbatim is a correctness requirement, not a formatting choice.

Figure — A single record advances through strict field checking then cross-field invariant checking to the scheduling queue; a violation at either stage branches it to a quarantine state that carries the ValidationError to the dead-letter queue, annotated with the guard that fired.

Integration Points #

This layer is a segment in a longer contract. Upstream, it consumes the ParsedSpotRecord stream that the parser emits; it treats those records as untrusted and re-derives nothing the parser guaranteed structurally. Downstream, every model that clears both gates is published to the scheduling queue, where the spot scheduling validation and rule engines stack applies rotation, conflict, and separation logic — most immediately detecting time-slot conflicts in traffic logs, which assumes every spot it receives is already daypart-aligned and duration-valid.

The message published on success is small, versioned, and carries the correlation ID so a spot can be traced end to end:

{
  "schema_version": "avstar-validate.v1",
  "correlation_id": "ingest-2026-183-0007",
  "spot_id": "AV-2026-183-0442",
  "client_code": "ACME",
  "campaign_id": "Q3-DRIVE-24",
  "spot_type": "COM",
  "daypart": "AFTD",
  "scheduled_airtime": "2026-07-03T22:30:00Z",
  "duration_seconds": 30,
  "billing_code": "RAW-77-Q3",
  "priority_tier": 2
}

A rejection publishes the ValidationOutcome shape to the dead-letter queue instead — same spot_id and correlation_id, plus the list of {field, value, constraint} triples — so triage never requires reading application logs.

Compliance & Audit Considerations #

Because this gate decides what is allowed to air, it is also a point that must be provable after the fact. Three rules are non-negotiable. First, every rejection is retained with its raw input and violated constraint; a dropped political or EAS-adjacent spot is a compliance incident, and FCC political-file reconstruction requires being able to show why a booked spot never entered the schedule. Second, the correlation ID minted at ingestion travels through validation unchanged, giving SOC 2 change-traceability an unbroken chain from source line to scheduling decision. Third, the validator process is least-privilege: it holds read access to the parsed stream and append-only rights to the audit log, under the model defined in security boundaries for traffic database access. Duration validation is the highest-risk surface here — a spot that airs even a fraction of a second off its booked length becomes a billing-reconciliation dispute — which is why the second boundary is enforced strictly at this gate rather than trusted from upstream.

Troubleshooting & Common Errors #

Related #

• Parsing Avion export formats — the structural extraction stage that produces the typed records this layer semantically validates.
• Validating spot durations against broadcast standards — a deep dive on mapping the duration constraint to frame-accurate network timing specs.
• Async batch processing for high-volume logs — how to stream validation over 10 GB+ drops under bounded memory with explicit backpressure.
• Standardizing billing codes across traffic systems — the canonicalization stage that owns billing-code mapping this validator only checks for presence.
• Avion & Avstar Ingestion Pipelines — the parent architecture that positions validation within the full ingestion-to-scheduling path.