Broadcast Traffic Scheduling Automation

Writing Custom Pydantic Validators for Avails

An avail — a saleable unit of commercial inventory in a specific break — carries constraints that no off-the-shelf type can express: its window must open before it closes, its daypart code must belong to a fixed vocabulary, its unit duration must land on a broadcast-legal boundary, and its remaining-units count must never exceed the units the break physically holds. This guide solves one exact task: writing Pydantic v2 field_validator and model_validator functions that enforce those avail rules at the ingestion boundary, so a malformed avail is rejected before it can be sold, scheduled, or billed. It is the validation procedure beneath Schema Validation with Pydantic for Traffic Data, part of Avion & Avstar Ingestion Pipelines. It matters for audit because an avail with an inverted window or an off-standard duration produces an as-run line that reconciliation cannot match to a contract, and an oversold avail double-books inventory that a public-file review will flag.

The core requirement is that validation be declarative, total, and fail-closed. Every field constraint lives on the model, every cross-field rule lives in a model_validator, and any violation raises rather than coerces — a duration that is close to legal is still illegal. This complements sibling work: the parent guide on validating spot durations against broadcast standards governs the individual spot, while this page governs the avail that spots are placed into, drawing its field definitions from the canonical traffic field data dictionary.

Prerequisites

  • Python 3.11+ — required for X | None union syntax, enum.StrEnum, and timezone-aware datetime.
  • Pydantic v2 — pin pydantic==2.7.1; the API here (field_validator, model_validator, ValidationInfo, mode="after") is v2-only and differs from v1’s @validator.
  • A daypart vocabulary — the station’s canonical daypart codes (for example AMDR, DAY, PRIME, LNGT) held in one place so the validator and the scheduler agree; these mirror the codes in Avails Mapping Strategies for Linear TV.
  • Broadcast-legal durations — the permitted spot lengths in seconds (10, 15, 30, 60, 90, 120) that an avail’s unit duration must match.
  • Read access to the raw avail feed — the Avion/Avstar export the models validate, delivered as a mounted file or an API pull.

Step-by-Step Implementation

An avail model validates in two passes: field validators normalize and range-check each field independently, then a model validator enforces cross-field invariants that no single field can see — window ordering, oversell, and daypart-duration compatibility. Validators run in a fixed order, so a field validator’s normalized output is what the model validator later reads.

Two-pass avail validation pipeline A raw avail record enters the model. In the first pass, three field validators run independently: one coerces the avail window start and end to timezone-aware UTC, one checks the daypart code against a fixed vocabulary and rejects unknown codes, and one checks the unit duration against the set of broadcast-legal spot lengths. Records failing any field validator are rejected immediately. Records that pass proceed to the second pass, a single model validator that sees all fields at once and enforces cross-field rules: the window must open strictly before it closes, remaining units must not exceed total units, and the duration must be permitted in that daypart. A record passing both passes becomes a validated Avail; any failure raises a ValidationError routed to a dead-letter queue. Raw avail Avion / Avstar Pass 1 · field_validator window → UTC reject naive datetimes daypart in vocab reject unknown code duration legal 10/15/30/60/90/120s Pass 2 · model_validator start < end remaining ≤ total duration ok in daypart Validated Avail ready to sell / schedule Dead-letter ValidationError + reason pass fail

Figure — Two-pass validation: field validators normalize and range-check each field, then a single model validator enforces the cross-field invariants — window ordering, oversell, and daypart-duration compatibility — before the avail is accepted.

Step 1 — Structured logging and the daypart vocabulary

Goal: emit greppable audit lines in the timestamp | level | module | spot_id shape and fix the vocabularies the validators check against. A daypart or duration that is not in the canonical set is a hard rejection.

python 
from __future__ import annotations

import logging
from datetime import datetime, timezone
from enum import StrEnum

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

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s | %(levelname)s | %(name)s | %(message)s",
)
logger = logging.getLogger("traffic.ingest.avail_validators")

# Broadcast-legal spot lengths in seconds; an avail unit must match one exactly.
LEGAL_DURATIONS_S: frozenset[int] = frozenset({10, 15, 30, 60, 90, 120})


class Daypart(StrEnum):
    AMDR = "amdr"     # AM drive
    DAY = "day"       # daytime
    PRIME = "prime"   # prime time
    LNGT = "lngt"     # late night

# Some dayparts forbid long-form units; prime never carries a 120s avail unit.
DAYPART_MAX_DURATION_S: dict[Daypart, int] = {
    Daypart.AMDR: 60,
    Daypart.DAY: 120,
    Daypart.PRIME: 90,
    Daypart.LNGT: 120,
}

Step 2 — Field validators for window, daypart, and duration

Goal: normalize and range-check each field on its own. A field_validator sees only its field, so it is the right place for coercion (naive datetime to UTC) and membership checks (daypart in vocabulary, duration in the legal set).

python 
class Avail(BaseModel):
    avail_id: str
    daypart: Daypart
    window_start: datetime
    window_end: datetime
    unit_duration_s: int = Field(gt=0)
    total_units: int = Field(ge=1)
    remaining_units: int = Field(ge=0)

    @field_validator("window_start", "window_end")
    @classmethod
    def _utc_aware(cls, v: datetime) -> datetime:
        # Naive datetimes silently corrupt window comparisons across DST.
        if v.tzinfo is None:
            raise ValueError("avail window bounds must be timezone-aware (UTC)")
        return v.astimezone(timezone.utc)

    @field_validator("unit_duration_s")
    @classmethod
    def _legal_duration(cls, v: int) -> int:
        # Fail closed on any length not in the broadcast-legal set.
        if v not in LEGAL_DURATIONS_S:
            raise ValueError(f"unit_duration_s={v} is not a broadcast-legal length")
        return v

Because daypart is typed as the Daypart enum, Pydantic already rejects an unknown code with a clear error — no explicit validator is needed for membership, only for the cross-field compatibility handled in Step 3. The legal-duration set mirrors the per-spot rule in validating spot durations against broadcast standards.

Step 3 — A model validator for cross-field invariants

Goal: enforce rules that need more than one field. An @model_validator(mode="after") runs once every field has been validated and coerced, so it can safely compare window_start to window_end, check oversell, and confirm the duration is permitted in this daypart.

python 
class Avail(BaseModel):  # continued — cross-field rule added to the model from Step 2
    @model_validator(mode="after")
    def _cross_field(self) -> "Avail":
        # 1. Window must open strictly before it closes.
        if self.window_start >= self.window_end:
            raise ValueError(
                f"avail window is inverted or zero-length: "
                f"{self.window_start.isoformat()} >= {self.window_end.isoformat()}"
            )
        # 2. Cannot have more units remaining than the break physically holds.
        if self.remaining_units > self.total_units:
            raise ValueError(
                f"oversold avail: remaining={self.remaining_units} "
                f"> total={self.total_units}"
            )
        # 3. Duration must be permitted in this daypart (e.g. no 120s in prime).
        ceiling = DAYPART_MAX_DURATION_S[self.daypart]
        if self.unit_duration_s > ceiling:
            raise ValueError(
                f"unit_duration_s={self.unit_duration_s} exceeds {ceiling}s "
                f"ceiling for daypart {self.daypart}"
            )
        logger.info("%s | avail validated daypart=%s dur=%ds units=%d/%d",
                    self.avail_id, self.daypart, self.unit_duration_s,
                    self.remaining_units, self.total_units)
        return self

Step 4 — A reusable validator with context

Goal: parameterize a rule with data unknown at class-definition time — for example a per-market daypart ceiling. A field_validator can read ValidationInfo.context, so the same model serves multiple markets without a subclass per market.

python 
class ContextualAvail(Avail):
    @field_validator("unit_duration_s")
    @classmethod
    def _market_ceiling(cls, v: int, info: ValidationInfo) -> int:
        # Context lets one model enforce a per-market override at parse time.
        ctx = info.context or {}
        market_cap: int | None = ctx.get("market_max_duration_s")
        if market_cap is not None and v > market_cap:
            raise ValueError(f"unit_duration_s={v} exceeds market cap {market_cap}s")
        return v


# Parse with per-call context; no subclass explosion per market.
avail = ContextualAvail.model_validate(
    {"avail_id": "AV-PRIME-1930-04", "daypart": "prime",
     "window_start": "2026-07-17T23:30:00+00:00",
     "window_end": "2026-07-17T23:32:00+00:00",
     "unit_duration_s": 30, "total_units": 6, "remaining_units": 4},
    context={"market_max_duration_s": 60},
)

A representative log line reads 2026-07-17T19:30:02+00:00 | INFO | traffic.ingest.avail_validators | AV-PRIME-1930-04 | avail validated daypart=prime dur=30s units=4/6.

Verification & Testing

Correctness rests on two properties: field validators reject bad values in isolation, and the model validator rejects bad combinations of individually valid values. Both are assertable against fixture data drawn from a real avail feed.

python 
from pydantic import ValidationError

base = {"avail_id": "AV-1", "daypart": "day",
        "window_start": "2026-07-17T14:00:00+00:00",
        "window_end": "2026-07-17T14:02:00+00:00",
        "unit_duration_s": 30, "total_units": 6, "remaining_units": 4}

# 1. Happy path parses.
ok = Avail.model_validate(base)
assert ok.remaining_units == 4

# 2. Inverted window fails in the model validator (fields individually valid).
try:
    Avail.model_validate({**base, "window_start": "2026-07-17T14:02:00+00:00",
                          "window_end": "2026-07-17T14:00:00+00:00"})
    assert False
except ValidationError as exc:
    assert "inverted" in str(exc)

# 3. Off-standard duration fails in the field validator.
try:
    Avail.model_validate({**base, "unit_duration_s": 45})
    assert False
except ValidationError as exc:
    assert "broadcast-legal" in str(exc)

# 4. Oversold avail fails.
try:
    Avail.model_validate({**base, "remaining_units": 9})
    assert False
except ValidationError as exc:
    assert "oversold" in str(exc)

Run this against a golden fixture of valid and invalid avails so any loosening of a validator — a stray daypart, a permitted 45-second unit — fails the build instead of admitting unsellable inventory. Assert on the reason substring, not just that a ValidationError was raised, so a rule that fails for the wrong cause is caught.

Edge Cases & Failure Handling

  • Zero-length or inverted window. A feed that emits window_start == window_end produces an avail no spot can fit. The model validator rejects it with an explicit “inverted or zero-length” reason rather than admitting a phantom avail; route the record to a dead-letter queue and reconcile against the source, since this is usually a timezone-truncation bug upstream.
  • Valid fields, invalid combination. An avail can have a legal duration and a known daypart yet still be illegal — a 120-second unit in a prime break whose ceiling is 90. Field validators pass such a record; only the model_validator catches it, which is exactly why cross-field rules must never be squeezed into a single-field validator where they cannot see the daypart.
  • DST boundary in the window. A window expressed in naive local time that straddles a spring-forward or fall-back distorts by an hour and can invert or balloon. The _utc_aware field validator rejects naive datetimes outright, and normalizing to UTC before comparison keeps the ordering check honest — the same discipline applied when mapping avails for linear TV.

FAQ

When should a rule be a field_validator versus a model_validator?

Use a field_validator when the rule needs only one field — coercing a datetime to UTC, checking a duration against a fixed set, normalizing a code. Use a model_validator(mode="after") when the rule compares two or more fields, such as window ordering or oversell, because a field validator cannot see the other fields. Putting a cross-field rule in a field validator is the single most common Pydantic mistake; it either fails silently or forces brittle ordering assumptions. The canonical field constraints guide draws the same line.

Why mode="after" instead of mode="before"?

mode="after" runs once every field has already been validated and coerced, so the model validator reads clean, typed values — the window bounds are already timezone-aware UTC, so the comparison is safe. mode="before" runs on the raw input dict before coercion, which is only appropriate for reshaping input, not for enforcing cross-field business rules. For avail invariants you almost always want after.

How do I reuse one model across markets with different ceilings?

Pass the per-market limit through model_validate(..., context={...}) and read it in a validator via ValidationInfo.context, as ContextualAvail does. That keeps one model definition instead of a subclass per market, and the limit travels with the parse call rather than being baked into the class. It composes cleanly with the ingestion flow in Schema Validation with Pydantic for Traffic Data.

Should a failed avail stop the whole batch?

No — reject the one record and continue. A ValidationError on a single avail should route that record to a dead-letter queue with its reason preserved, while valid avails proceed, so one bad row never blocks a night’s ingestion. This mirrors the per-record isolation used throughout Avion & Avstar Ingestion Pipelines; only a systemic failure, caught by a circuit breaker, should halt the batch.

Related content

Up one levelSchema Validation with Pydantic for Traffic Data RelatedValidating Spot Durations Against Broadcast Standards