Broadcast Traffic Scheduling Automation

Validating Canonical Field Constraints in Python

A field data dictionary is only as trustworthy as the code that enforces it. A duration_ms documented as “greater than zero” that no validator checks is a suggestion, not a constraint — and the first zero-duration airing to slip through posts a phantom spot into billing. This guide solves one exact task: turning the constraint column of The Canonical Traffic Field Data Dictionary into runnable Pydantic v2 validators, then aggregating every violation into a single constraint-violation report a traffic engineer can act on. It sits under that dictionary within Broadcast Traffic Architecture & Taxonomy, and it matters because the constraints are exactly the properties an FCC public-inspection review and a SOC 2 reproducibility audit assume were enforced at ingestion — a naive datetime or a missing sponsorship ID that reaches the archive is a compliance finding, not a bug ticket.

The governing principle is fail-closed with full accounting. A record that violates a constraint is never coerced into looking valid; it is rejected with a precise, machine-readable reason and collected into a report, so a whole batch is validated in one pass rather than dying on its first bad row. That is the same boundary-validation discipline the Pydantic validator pattern applies across ingestion and the spot schema assumes for identity fields.

Prerequisites

  • Python 3.11+ — required for X | None unions, datetime.now(timezone.utc), and the exception-group-friendly error handling used below.
  • Pydantic v2 — pin it exactly (pydantic==2.7.1); this guide uses field_validator, model_validator, and ValidationError.errors(), whose shapes changed from v1.
  • The constraint table — the type, constraint, and nullability columns from the canonical data dictionary; this guide enforces them, so any change there changes the validators here.
  • Canonical field names already applied — run vendor field mapping first so records arrive under stable snake_case names; this validator checks constraints, not field structure.
  • A writable report location — a path or object-store prefix where the constraint-violation report is written for the traffic desk.

Step-by-Step Implementation

Validation runs as a pure function over each candidate record: construct the Pydantic model, catch the structured ValidationError, and translate it into typed violations. A batch runner accumulates violations across all rows and emits one report, so validation is a single auditable pass rather than a stop-on-first-error crash.

Constraint validation and reporting flow A candidate record enters the constraint validator, which attempts to construct the CanonicalSpot Pydantic model. Field validators check timezone-aware datetimes, positive duration, non-negative rate, and pattern constraints; a model validator checks cross-field rules like sponsored requiring a sponsorship id. If construction succeeds the record is emitted as valid. If it raises a ValidationError, each error is translated into a typed violation with a field, code, and message, and appended to a violation report. The batch runner processes every record before writing one aggregated report, so validation never stops on the first bad row. Candidate record Constraint validator field: tz-aware · gt 0 · ge 0 field: pattern · enum · length model: sponsored ⇒ sponsorship_id model: aired_at ≥ scheduled_at CanonicalSpot(...) ok? Valid record emitted downstream Violation report field · code · message one row per breach yes no batch runner · next record · never stops on first breach

Figure — Constraint validation: each record is constructed against the Pydantic model, valid ones pass through, and each violation is translated into a typed report row while the batch runner continues to the next record.

Step 1 — Model the constraints and a typed violation

Goal: encode the dictionary’s per-field constraints declaratively on the model and define the violation record the report is built from. Declarative constraints (gt, ge, pattern) keep the rules readable and colocated with the fields they govern.

python 
from __future__ import annotations

import logging
from dataclasses import dataclass
from datetime import datetime, timezone
from typing import Optional

from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator

# Traffic-ops structured logging: timestamp | level | module | spot_id
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.canonical.constraint_validator")


@dataclass(frozen=True)
class Violation:
    spot_id: str
    field: str
    code: str          # stable ERR_* code for dashboards
    message: str


class CanonicalSpot(BaseModel):
    """Constraint-bearing subset of the canonical dictionary for validation."""

    model_config = {"extra": "forbid"}

    spot_id: str = Field(min_length=1)
    house_number: str = Field(pattern=r"^[A-Z0-9-]{3,32}$")
    order_id: str = Field(min_length=1)
    sponsor_id: str = Field(min_length=1)
    isci: Optional[str] = Field(default=None, pattern=r"^[A-Z0-9]{8,12}$")
    scheduled_at: datetime
    aired_at: Optional[datetime] = None
    duration_ms: int = Field(gt=0)            # zero-duration = phantom spot
    rate_cents: int = Field(ge=0)             # integer cents, never negative
    billing_code: str = Field(pattern=r"^[A-Z0-9]{4,}$")
    advertiser_category: str = Field(min_length=1)
    is_sponsored: bool = False
    sponsorship_id: Optional[str] = None

Step 2 — Enforce constraints Pydantic types cannot express

Goal: add the field- and cross-field rules the dictionary requires but a plain type annotation cannot — timezone-awareness on datetimes and the FCC-driven rule that a sponsored airing must carry a sponsorship ID. These run automatically during model construction.

python 
class CanonicalSpot(BaseModel):  # continued — validators on the model from Step 1
    @field_validator("scheduled_at", "aired_at")
    @classmethod
    def _utc_aware(cls, v: Optional[datetime]) -> Optional[datetime]:
        # A naive datetime crossing a DST boundary silently shifts an airing an
        # hour, corrupting separation and reconciliation. Reject it outright.
        if v is None:
            return v
        if v.tzinfo is None:
            raise ValueError("must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)

    @model_validator(mode="after")
    def _cross_field(self) -> "CanonicalSpot":
        # FCC Section 317: sponsored content must carry its sponsorship id.
        if self.is_sponsored and not self.sponsorship_id:
            raise ValueError("is_sponsored requires sponsorship_id")
        # An airing cannot occur before it was scheduled.
        if self.aired_at and self.aired_at < self.scheduled_at:
            raise ValueError("aired_at precedes scheduled_at")
        return self

Step 3 — Translate a ValidationError into typed violations

Goal: convert Pydantic’s structured error list into stable Violation rows with an ERR_* code, so a report is dashboard-ready rather than a wall of raw tracebacks. Mapping Pydantic error types to codes keeps the report vocabulary stable even if a message string changes.

python 
# Pydantic error 'type' -> stable ERR_* code for the dictionary's constraints.
_CODE_BY_TYPE: dict[str, str] = {
    "greater_than": "ERR_DURATION_INVALID",
    "greater_than_equal": "ERR_RATE_NEGATIVE",
    "string_pattern_mismatch": "ERR_PATTERN",
    "missing": "ERR_REQUIRED_MISSING",
    "extra_forbidden": "ERR_UNKNOWN_FIELD",
    "value_error": "ERR_RULE_VIOLATION",   # our raised ValueErrors land here
}


def _violations_from(spot_id: str, exc: ValidationError) -> list[Violation]:
    out: list[Violation] = []
    for err in exc.errors():
        field = ".".join(str(p) for p in err["loc"]) or "<model>"
        code = _CODE_BY_TYPE.get(err["type"], "ERR_CONSTRAINT")
        out.append(Violation(spot_id, field, code, err["msg"]))
    return out

Step 4 — Validate one record, fail closed

Goal: wrap construction so a valid record returns the model and an invalid one returns its violations — never a half-populated object. The function is pure, so the same record always produces the same verdict, the property an audit replay depends on.

python 
@dataclass(frozen=True)
class ValidationOutcome:
    spot: Optional[CanonicalSpot]
    violations: list[Violation]


def validate_record(raw: dict[str, object]) -> ValidationOutcome:
    # spot_id is the audit key; fall back so even a malformed row is traceable.
    spot_id = str(raw.get("spot_id", "UNKNOWN"))
    try:
        spot = CanonicalSpot(**raw)  # type: ignore[arg-type]
        logger.info("%s | validated OK", spot_id)
        return ValidationOutcome(spot=spot, violations=[])
    except ValidationError as exc:
        violations = _violations_from(spot_id, exc)
        for v in violations:
            logger.warning("%s | %s on %s: %s", v.spot_id, v.code, v.field, v.message)
        return ValidationOutcome(spot=None, violations=violations)

Step 5 — Run a batch and emit one constraint-violation report

Goal: validate every record before writing a single aggregated report, so one bad row never masks the next. The report groups violations by ERR_* code so the traffic desk sees the shape of the failure at a glance.

python 
from collections import Counter
from pathlib import Path
import json


def validate_batch(records: list[dict[str, object]], report_dir: Path) -> list[CanonicalSpot]:
    valid: list[CanonicalSpot] = []
    all_violations: list[Violation] = []
    for raw in records:
        outcome = validate_record(raw)
        if outcome.spot is not None:
            valid.append(outcome.spot)
        all_violations.extend(outcome.violations)

    report_dir.mkdir(parents=True, exist_ok=True)
    report = report_dir / "constraint_violations.json"
    by_code = Counter(v.code for v in all_violations)
    report.write_text(json.dumps({
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "records_checked": len(records),
        "records_valid": len(valid),
        "violations_by_code": dict(by_code),
        "violations": [v.__dict__ for v in all_violations],
    }, indent=2))
    logger.info("SYSTEM | batch: %d/%d valid, %d violations -> %s",
                len(valid), len(records), len(all_violations), report)
    return valid

Expected summary log line: 2026-07-17T10:05:14 | INFO | traffic.canonical.constraint_validator | SYSTEM | batch: 2/4 valid, 3 violations -> reports/constraint_violations.json.

Verification & Testing

Two properties define correctness: a record that satisfies every constraint validates, and each constraint breach produces exactly the expected ERR_* code — a bad row never returns a usable model. Both are assertable against fixtures that exercise the boundaries the dictionary defines.

python 
good = {
    "spot_id": "SP-1", "house_number": "RCPRIME30", "order_id": "ORD-1",
    "sponsor_id": "SPN-1", "scheduled_at": datetime(2026, 7, 17, 23, 30, tzinfo=timezone.utc),
    "duration_ms": 30000, "rate_cents": 285000, "billing_code": "RCPRIME30",
    "advertiser_category": "auto",
}
assert validate_record(good).spot is not None          # all constraints satisfied

# Zero duration -> phantom-spot guard fires with the stable code.
bad_dur = validate_record({**good, "duration_ms": 0})
assert bad_dur.spot is None
assert any(v.code == "ERR_DURATION_INVALID" for v in bad_dur.violations)

# Naive datetime -> rejected as a DST-corruption risk.
naive = validate_record({**good, "scheduled_at": datetime(2026, 7, 17, 23, 30)})
assert any(v.field == "scheduled_at" for v in naive.violations)

# Sponsored without a sponsorship id -> FCC cross-field rule fails closed.
sponsored = validate_record({**good, "is_sponsored": True})
assert any(v.code == "ERR_RULE_VIOLATION" for v in sponsored.violations)

Because validate_record is pure, a fixture that fails today fails identically on replay during an audit. Run the suite in CI against a golden set of valid and invalid records so a loosened constraint — say gt=0 weakened to ge=0 on duration_ms — fails the build instead of quietly admitting phantom spots into the schedule.

Edge Cases & Failure Handling

  • Multiple violations on one record. A row can breach several constraints at once — a naive scheduled_at and a negative rate_cents. Pydantic collects them all in one ValidationError, and _violations_from emits one report row per breach, so the traffic desk fixes every problem in a single pass rather than rediscovering the next one after each re-run.
  • An unknown field slips through mapping. With extra="forbid", a stray column that vendor field mapping failed to catch surfaces here as ERR_UNKNOWN_FIELD rather than being silently absorbed. Resolve it in the rename map upstream; never relax extra to make the error disappear, which is how a mis-typed value reaches the canonical layer.
  • A constraint that genuinely needs to change. Standard lengths drift and categories are added over time. When a constraint must move, change it in the canonical data dictionary first and update these validators to match in the same commit, so the documented contract and the enforced one never diverge — a validator stricter or looser than the published dictionary is its own kind of audit finding.

FAQ

Why aggregate violations into a report instead of raising on the first bad record?

Because a nightly export carries thousands of rows, and stopping on the first breach means the traffic desk fixes one problem, re-runs, hits the next, and repeats for hours. Validating the whole batch and emitting one grouped report surfaces the full shape of the failure at once. The record-level fail-closed guarantee still holds — no invalid row ever returns a usable model — it is only the batch that continues, mirroring the boundary discipline in schema validation with Pydantic.

Should I use field_validator or model_validator for a given rule?

Use field_validator for anything about a single field’s own value — timezone-awareness, pattern, range. Use model_validator(mode="after") for rules that read two or more fields, like sponsored requiring a sponsorship id, or aired_at not preceding scheduled_at. Keeping single-field checks off the model validator makes each error point at the exact field, which is what makes the constraint-violation report actionable rather than a vague “record invalid”.

How do I keep the ERR_ codes stable when Pydantic messages change?

Map Pydantic’s error type (a stable machine string like greater_than or string_pattern_mismatch) to your own ERR_* code, never the human-readable msg. The message text can change between Pydantic releases, but the type is part of its contract, so a dashboard keyed on your codes survives library upgrades. The codes here align with the troubleshooting table in the canonical data dictionary so one vocabulary spans validation and operations.

Where does constraint validation sit relative to vendor field mapping?

Mapping runs first and answers “which column is this field”; validation runs second and answers “does this field’s value obey the dictionary”. Run them in that order — vendor field mapping hands this validator records already under canonical names, so a failure here is unambiguously a constraint breach and not a naming problem. Combining the two passes would blur which layer a failure came from.

Can the same validator run in ingestion and in as-run reconciliation?

Yes, and it should — a single shared model is the point of a canonical dictionary. The same CanonicalSpot constraints that guard ingestion also guard the records fed into as-run reconciliation, so a spot that was valid on the way in is validated identically on the way to billing. Importing one model everywhere prevents two subsystems from disagreeing about what “valid” means.

Related content

Up one levelThe Canonical Traffic Field Data Dictionary RelatedHow to Map Vendor Fields to Canonical Names