Avstar API Authentication and Rate Limits

When an ingestion job pushes a night’s worth of commercial schedules into Avstar, the failure that hurts most is not a malformed spot — it is a dropped or throttled session that silently strands half a traffic log between the ingestion layer and the automation platform. Without a disciplined authentication and pacing layer, tokens expire mid-batch, 429 Too Many Requests responses cascade into retry storms, and partially-submitted logs force manual reconciliation the next morning. This is the control plane of the Avion & Avstar Ingestion Pipelines: the subsystem that establishes a secure, quota-aware session, negotiates API capacity, and enforces strict error boundaries before any normalized traffic record is handed to the scheduling engine. It is written for broadcast traffic managers who need to understand why submissions stall, and for Python automation builders who need a deployable client that never loses a spot to an expired token.

Authentication and rate limiting are not incidental plumbing here — they are the guarantee that a submitted schedule is either fully accepted or cleanly rejected, with an audit trail either way. Everything upstream (parsing, validation) produces records; this layer is the last deterministic gate before those records leave your process boundary.

Concept & Data Model #

Avstar authentication is a stateful, decoupled process, not a per-request credential injection. Treating it as state — a token with an issue time, a lifetime, and a refresh policy — is what lets a long-running batch survive token expiry without dropping in-flight work. Three entities model the domain:

• Credential — the long-lived client_id / client_secret pair (or scoped API key) held outside application code in an environment variable, HashiCorp Vault, or AWS Secrets Manager. Credentials are exchanged for tokens; they are never sent on data requests.
• Access Token — a short-lived bearer credential returned by the OAuth 2.0 client-credentials grant. It carries an expires_in lifetime and an implicit issued_at. The token, not the credential, authorizes each traffic:write or inventory:read call.
• Rate Budget — the server-advertised request allowance for the current window, surfaced through RateLimit-Remaining and RateLimit-Reset response headers. The client mirrors this budget locally so it can pace itself instead of discovering the ceiling by hitting 429.

The token moves through a small state machine. It begins absent, becomes valid after a successful grant, is treated as stale once it enters a configurable skew window before hard expiry, and returns to absent on a 401. The rate budget runs in parallel: open while remaining requests exist, throttled on a 429 until RateLimit-Reset, then open again.

The fields the client actually persists between requests are narrow and typed. The token manager guarantees these before any submission is attempted.

Records arriving at this layer are already normalized against the canonical spot schema and metadata, so the auth client never inspects business fields — it only carries a validated spot_id through for logging and idempotency.

Implementation Approach #

Two design decisions dominate this layer, and both trade a little complexity for a lot of resilience.

Cache-and-refresh versus fetch-per-request. Requesting a fresh token on every call is simple and stateless, but at broadcast batch volumes it doubles request count and burns rate budget on oauth/token round trips. The production choice is a cached token with a proactive skew window: refresh 120 seconds before hard expiry so no in-flight request ever races the expiry boundary. The token manager is injected as a shared dependency into async worker pools so a single refresh serves every concurrent submission, rather than each worker triggering its own grant.

Client-side pacing versus reactive backoff. Relying only on 429 responses to discover the ceiling means you always overshoot by exactly the burst that tripped it — expensive when a sports overrun compresses a whole daypart’s submissions into one window. The stronger approach pairs a local token-bucket limiter (sized to the server-advertised budget) with reactive exponential backoff plus jitter as a safety net. The bucket prevents most 429s; the backoff absorbs the rest without synchronized retry storms across distributed workers.

These choices connect directly to sibling subsystems. Token exchange sits behind the same trust boundary described in security boundaries for traffic database access, and scope selection follows the least-privilege model in role-based access for traffic APIs. For the batch mechanics that this client paces, see async batch processing for high-volume logs; for the timeout recovery that complements refresh, see handling Avstar session timeouts in Python.

Production Python Implementation #

The module below is deployable as written. It composes a TokenManager (state machine over the access token) with a RateBudget (token-bucket limiter that respects server headers) into an AvstarClient that submits a normalized spot and never loses it to an expired token or a throttle. Every operational event emits a structured log line in the traffic-ops format timestamp | level | module | spot_id, so submissions are traceable in the same log stream as the rest of the pipeline.

from __future__ import annotations

import asyncio
import logging
import random
from datetime import datetime, timedelta, timezone

import httpx
from pydantic import BaseModel, Field, field_validator

logger = logging.getLogger("avstar.auth")

# Traffic-ops log format: timestamp | level | module | spot_id
_LOG_FMT = "%(asctime)s | %(levelname)s | %(name)s | %(spot_id)s | %(message)s"


def _configure_logging() -> None:
    handler = logging.StreamHandler()
    handler.setFormatter(logging.Formatter(_LOG_FMT, datefmt="%Y-%m-%dT%H:%M:%SZ"))
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)


def _log(level: int, msg: str, spot_id: str = "-") -> None:
    # spot_id is a sentinel ('-') for session-level events, real UUID for submissions.
    logger.log(level, msg, extra={"spot_id": spot_id})


def _utcnow() -> datetime:
    return datetime.now(timezone.utc)


class AvstarTokenResponse(BaseModel):
    """Typed view of the OAuth client-credentials grant response."""

    access_token: str = Field(..., min_length=1)
    token_type: str = Field(default="Bearer")
    expires_in: int = Field(..., gt=0)  # seconds until hard expiry
    scope: str = Field(default="traffic:write inventory:read")
    issued_at: datetime = Field(default_factory=_utcnow)

    @field_validator("issued_at")
    @classmethod
    def _must_be_aware(cls, v: datetime) -> datetime:
        # Naive timestamps silently corrupt expiry math across DST/timezones.
        if v.tzinfo is None:
            raise ValueError("issued_at must be timezone-aware UTC")
        return v

    def hard_expiry(self) -> datetime:
        return self.issued_at + timedelta(seconds=self.expires_in)


class TokenManager:
    """Owns the access-token state machine: absent -> valid -> stale -> refresh."""

    def __init__(self, client_id: str, client_secret: str, base_url: str,
                 skew_seconds: int = 120) -> None:
        self._client_id = client_id
        self._client_secret = client_secret
        self._base_url = base_url.rstrip("/")
        self._skew = timedelta(seconds=skew_seconds)
        self._token: AvstarTokenResponse | None = None
        self._lock = asyncio.Lock()  # one refresh serves all concurrent workers

    def _is_valid(self) -> bool:
        # A token is reusable only until it enters the skew window before expiry.
        return bool(self._token) and _utcnow() < self._token.hard_expiry() - self._skew

    async def get_valid_token(self) -> str:
        if self._is_valid():
            return self._token.access_token  # type: ignore[union-attr]
        async with self._lock:
            if self._is_valid():  # double-check: another worker may have refreshed
                return self._token.access_token  # type: ignore[union-attr]
            await self._refresh()
            return self._token.access_token  # type: ignore[union-attr]

    async def invalidate(self) -> None:
        # Called on a 401 so the next request forces a fresh grant.
        async with self._lock:
            self._token = None
            _log(logging.WARNING, "token invalidated after 401")

    async def _refresh(self) -> None:
        async with httpx.AsyncClient(timeout=10.0) as client:
            resp = await client.post(
                f"{self._base_url}/oauth/token",
                data={
                    "grant_type": "client_credentials",
                    "client_id": self._client_id,
                    "client_secret": self._client_secret,
                    "scope": "traffic:write inventory:read",  # least privilege
                },
            )
            resp.raise_for_status()
            self._token = AvstarTokenResponse(**resp.json())
        _log(logging.INFO, f"token refreshed, ttl={self._token.expires_in}s")


class RateBudget:
    """Token-bucket limiter mirroring the server-advertised rate window."""

    def __init__(self, capacity: int = 60, refill_per_sec: float = 1.0) -> None:
        self._capacity = capacity
        self._tokens = float(capacity)
        self._refill = refill_per_sec
        self._reset_at: datetime | None = None
        self._updated = _utcnow()
        self._lock = asyncio.Lock()

    def sync_from_headers(self, headers: httpx.Headers) -> None:
        # Trust the server's numbers over our estimate whenever they are present.
        remaining = headers.get("RateLimit-Remaining")
        reset = headers.get("RateLimit-Reset")
        if remaining is not None:
            self._tokens = float(int(remaining))
        if reset is not None:
            self._reset_at = _utcnow() + timedelta(seconds=int(reset))

    async def acquire(self, spot_id: str) -> None:
        async with self._lock:
            now = _utcnow()
            elapsed = (now - self._updated).total_seconds()
            self._tokens = min(self._capacity, self._tokens + elapsed * self._refill)
            self._updated = now
            if self._tokens < 1.0:
                wait = 1.0 / self._refill
                _log(logging.INFO, f"rate budget exhausted, pacing {wait:.1f}s",
                     spot_id=spot_id)
                await asyncio.sleep(wait)
                self._tokens = 1.0
            self._tokens -= 1.0


class SpotSubmission(BaseModel):
    spot_id: str = Field(..., min_length=1)  # UUIDv5 idempotency key
    payload: dict


class AvstarClient:
    """Submits a normalized spot with token refresh, pacing, and bounded retry."""

    _RETRYABLE = {429, 502, 503, 504}

    def __init__(self, tokens: TokenManager, budget: RateBudget, base_url: str,
                 max_retries: int = 5) -> None:
        self._tokens = tokens
        self._budget = budget
        self._base_url = base_url.rstrip("/")
        self._max_retries = max_retries

    async def submit_spot(self, spot: SpotSubmission) -> dict:
        attempt = 0
        while True:
            await self._budget.acquire(spot.spot_id)
            token = await self._tokens.get_valid_token()
            async with httpx.AsyncClient(timeout=15.0) as client:
                resp = await client.post(
                    f"{self._base_url}/traffic/logs",
                    headers={"Authorization": f"Bearer {token}"},
                    json=spot.payload,
                )
            self._budget.sync_from_headers(resp.headers)

            if resp.status_code == 401:
                await self._tokens.invalidate()  # stale token -> refresh, do not count as retry
                continue
            if resp.status_code in self._RETRYABLE:
                attempt += 1
                if attempt > self._max_retries:
                    _log(logging.ERROR, f"submission abandoned after {attempt} retries",
                         spot_id=spot.spot_id)
                    resp.raise_for_status()
                backoff = min(2 ** attempt, 30) + random.uniform(0, 1)  # jitter
                _log(logging.WARNING,
                     f"status={resp.status_code} retry {attempt} in {backoff:.1f}s",
                     spot_id=spot.spot_id)
                await asyncio.sleep(backoff)
                continue

            resp.raise_for_status()
            _log(logging.INFO, "spot accepted by Avstar", spot_id=spot.spot_id)
            return resp.json()

The client separates concerns cleanly: TokenManager never knows about spots, RateBudget never knows about auth, and AvstarClient orchestrates both around a single idempotent submission. A 401 triggers re-authentication without consuming a retry, so an expired token can never exhaust the retry budget the way a genuine 503 would.

Validation & Edge Cases #

Broadcast timing makes several edge cases mandatory rather than optional:

• Clock skew and DST boundaries. Expiry math on a naive datetime drifts by an hour twice a year and by seconds continuously. The AvstarTokenResponse validator rejects any non-UTC-aware issued_at, and the 120-second skew window absorbs residual client/server clock disagreement.
• Sports overruns compressing submissions. When a live event runs long, a whole daypart of make-good and revised placements can flush into one narrow window. The token-bucket ceiling paces these instead of firing every submission at once and collecting a wall of 429s.
• Zero-remaining budget with a distant reset. If RateLimit-Remaining is 0 and RateLimit-Reset is 40 seconds out, the client must wait, not spin. sync_from_headers records the reset so pacing reflects the server’s real window rather than the local estimate.
• Refresh stampede under concurrency. Dozens of workers hitting a stale token simultaneously must not each request a new grant. The asyncio.Lock plus double-checked validity ensures exactly one refresh per expiry, shared by all.
• Idempotent replay after partial failure. Because spot_id is a UUIDv5 idempotency key validated against the Pydantic traffic-data schema, a retried submission after a network drop is safe: Avstar deduplicates on the key rather than double-placing the spot.

Integration Points #

This layer is the handoff seam between ingestion and scheduling. Upstream, parsing Avion export formats produces raw records and schema validation with Pydantic for traffic data guarantees their shape; only then does a record reach submit_spot. Downstream, accepted logs feed the deterministic spot scheduling validation and rule engines, which assume every record they receive was authenticated, paced, and acknowledged.

The submission contract the client enforces is intentionally small:

{
  "spot_id": "8c1e0e2a-6b3f-5c7a-9d21-2f0a4b6c8e10",
  "avail_id": "AV-20260703-PRIME-014",
  "advertiser_id": "ADV-4471",
  "duration_seconds": 30,
  "billing_code": "AGY-NAT-30-Q3",
  "daypart": "PRIME"
}

The billing_code travels through this boundary untouched but must already conform to the conventions in standardizing billing codes across traffic systems, because reconciliation later joins on it. A response returns the Avstar-assigned log position and an acknowledgement receipt, which the pipeline archives immutably for audit.

Compliance & Audit Considerations #

Authentication and pacing are audit surfaces, not just reliability features. Three rules apply:

• Least-privilege scope. The client-credentials grant requests only traffic:write inventory:read. A submission client that cannot delete or administer inventory limits the blast radius of a leaked token and satisfies SOC 2 access-control expectations.
• Immutable submission ledger. Every accepted spot must produce an append-only record — spot_id, token scope, request timestamp, Avstar receipt ID, and response status — written to a WORM-style store. This ledger is the primary evidence in billing reconciliation and, where political advertising is involved, in demonstrating timely placement against FCC political file obligations. The structured spot_id-keyed log lines emitted above are the raw material for that ledger.
• No credential in the log stream. The traffic-ops log format carries spot_id, status, and timing — never the token or secret. Redaction at the logging boundary is a hard requirement, since these logs are retained for years for audit reconstruction.

Troubleshooting & Common Errors #

Related #

• Handling Avstar Session Timeouts in Python — deterministic timeout recovery and connection-pool recycling that complements proactive token refresh.
• Async Batch Processing for High-Volume Logs — the bounded-concurrency batch mechanics this client paces against the rate budget.
• Schema Validation with Pydantic for Traffic Data — the Pydantic validator gate that guarantees every record is well-formed before submission.
• Security Boundaries for Traffic Database Access — the trust boundary and least-privilege model that credential exchange sits behind.