Async Batch Processing for Film/TV Cost Ingestion: Architecture, Compliance, and Production-Ready Python

A single principal-photography day can emit thousands of cost records across departmental purchase orders, vendor invoices, payroll runs, and location settlements — arriving from studio accounting portals, equipment-rental APIs, and SFTP flat-file drops at unpredictable cadences. A synchronous pipeline that blocks the ledger process on the slowest of those sources turns the nightly close into a serial queue: one throttled vendor API stalls guild payroll that is already inside its remittance window, month-end reconciliation slips, and a completion-guarantor’s cost-to-complete report goes stale before it is even generated. Async batch processing is the layer that removes that coupling. By fetching every source concurrently under a bounded worker pool, validating each record against a strict monetary schema, and routing failures deterministically instead of aborting the run, the ingestion tier keeps throughput high without ever sacrificing the audit trail a bond lender expects. This guide specifies that layer — the concurrency model, the production-grade Python, the guild-rate enforcement, and the dead-letter mechanics — for production accountants, line producers, and the automation engineers who have to make a chaotic data day reproducible.

Prerequisites and Expected Inputs

The pipeline described here targets Python 3.11+, both for zoneinfo in the standard library and for modern union-type syntax. It leans on a small, deliberate dependency set: asyncio and aiohttp for non-blocking I/O and connection-pooled HTTP; Pydantic v2 for boundary schema validation using model_validate and field_validator; and the standard-library decimal, hashlib, and zoneinfo modules for currency-safe arithmetic, deterministic audit hashing, and timezone-aware timestamps. Larger columnar normalization — de-duplicating and reshaping wide vendor exports before validation — is best handled with polars, and the append-only ledger and dead-letter tables are typically fronted by SQLAlchemy 2.0. Never use float for monetary values: a single fractional cent, compounded across tens of thousands of daily transactions, becomes exactly the variance a guarantor will ask you to explain.

The service expects three input shapes, each pulled by a different transport: RESTful JSON payloads from vendor and post-production APIs, CSV or flat-file exports delivered over SFTP, and structured payroll edits exported from an accounting platform such as Entertainment Partners (EP) or Showbiz. Every record carries, at minimum, an untrusted transaction identifier, a monetary amount, an ISO currency code, a general-ledger (GL) code, and an ingestion-time reference date. Async batch processing sits directly downstream of transport and directly upstream of the ledger: the deterministic entry and normalization semantics it depends on are specified in CSV & API Sync Pipelines, and the boundary contracts it enforces are drawn from Schema Validation & Error Handling. This page is one subsystem of the broader Cost Ingestion & Data Parsing Workflows reference architecture, which every record must ultimately satisfy before it reaches the general ledger.

Architecture: The Non-Blocking Ingestion Path

The core design decision is to treat each source as an independent, cancellable task scheduled on a single event loop, rather than a step in a serial script. The event loop yields control whenever a fetch is waiting on the network, so a high-priority guild payroll file is never held hostage by a slow vendor CSV. Concurrency is deliberately bounded: an asyncio.Semaphore caps the number of in-flight connections so the pipeline respects vendor rate limits and never exhausts the host’s file descriptors or the connection pool during a heavy month-end load. Raw batches return from the pool, are split into fixed-size chunks to cap peak memory, and each chunk is validated in place before its valid records are routed to the ledger and its failures to a dead-letter queue.

Two structural rules make this durable. First, back-pressure is explicit, not incidental: the semaphore and the fixed chunk size are the two knobs that keep memory and connection count bounded regardless of how large a wrap-day batch turns out to be. Second, failure is a routing outcome, not a halt — an infrastructure error on one source (a 503 from a European rental API during peak wrap) is caught, the source is skipped, and the remaining sources still commit. Nothing about one vendor’s outage is allowed to corrupt or delay another vendor’s clean records. This is the same non-blocking substrate that CSV & API Sync Pipelines rely on to absorb payroll-provider rate limits, and that EP/Showbiz Sync Parsing uses to reconcile legacy flat files against live API payloads without stalling.

The sequence below traces how a semaphore-bounded worker pool acquires a slot, fetches each source, chunks the batch, validates records, and splits the results between the ledger and the dead-letter queue.

Semaphore-bounded worker pool acquiring a slot, fetching, chunking, validating, and splitting output A sequence over six lifelines — Orchestrator, Semaphore, Worker pool, Vendor sources, Ledger, and Dead-letter queue. The orchestrator acquires the semaphore capped at ten; the semaphore grants a slot to the worker pool, which fetches vendor data and receives a raw batch. Inside a loop run once per chunk, the worker validates each record against CostRecord, routes valid records to the ledger, and routes failed records with their payload and SHA-256 hash to the dead-letter queue. The worker then releases the slot back to the semaphore. Orchestrator Semaphore Worker pool Vendor sources Ledger Dead-letter queue acquire · cap 10 slot granted fetch vendor data raw batch loop · per chunk (batch_size) validate against CostRecord route valid records route failed · payload + SHA-256 hash release slot
Semaphore-bounded worker pool: acquire a bounded slot, fetch each source, then loop per chunk — validate against CostRecord, route valid records to the ledger and failures to the dead-letter queue — before releasing the slot.

Core Implementation: A Semaphore-Bounded Ingestion Pipeline

The reference implementation below composes four responsibilities that stay strictly separated: the schema that defines a valid cost record, the bounded fetcher that pulls a source, the per-chunk validator that fingerprints failures, and the orchestrator that ties them together. Read it as four steps.

Step 1 — the boundary schema. CostRecord is the single contract every record must satisfy before the ledger will accept it. Monetary precision is protected by typing amount as Decimal, and a field_validator rejects negative amounts so that a reversal is forced through its own separate workflow rather than silently netting against a charge. The optional union_category is where guild-rate routing hooks in later.

Step 2 — the bounded fetcher. fetch_vendor_data acquires the semaphore before it opens a connection, so the cap on concurrency is enforced at the exact moment a socket would otherwise be created. raise_for_status() converts an HTTP error into an exception the orchestrator can route rather than a silently empty batch.

Step 3 — validate and fingerprint. process_batch walks a chunk, attempts CostRecord.model_validate on each row, and on failure preserves the original payload verbatim alongside its ValidationError detail and a deterministic SHA-256 fingerprint. The payload is never mutated in place — the record that failed must be reconstructable byte-for-byte during an audit.

Step 4 — orchestrate under back-pressure. run_async_ingestion_pipeline dispatches every source concurrently with asyncio.gather(..., return_exceptions=True), so one source raising does not cancel the others; it then chunks each successful batch to cap memory and routes the two output streams.

import asyncio
import hashlib
import json
from datetime import datetime
from decimal import Decimal
from typing import Any
from zoneinfo import ZoneInfo

import aiohttp
from pydantic import BaseModel, ValidationError, field_validator

# Production accounting is anchored to a studio's operating timezone, not a bare UTC offset.
PRODUCTION_TZ = ZoneInfo("America/Los_Angeles")


# Strict cost record schema aligned with studio GL and union compliance requirements.
class CostRecord(BaseModel):
    transaction_id: str
    department_code: str
    amount: Decimal  # Decimal preserves exact monetary precision; never use float for money
    currency: str
    gl_code: str
    union_category: str | None = None
    timestamp: datetime

    @field_validator("amount")
    @classmethod
    def enforce_non_negative(cls, v: Decimal) -> Decimal:
        if v < 0:
            raise ValueError("Negative amounts require a separate reversal workflow per bond standards")
        return v


async def fetch_vendor_data(
    session: aiohttp.ClientSession, url: str, semaphore: asyncio.Semaphore
) -> list[dict[str, Any]]:
    async with semaphore:  # cap enforced exactly where a socket would be opened
        async with session.get(url) as response:
            response.raise_for_status()
            return await response.json()


def _fingerprint(record: dict[str, Any]) -> str:
    # Canonical, sorted serialization yields a deterministic, auditable hash.
    canonical = json.dumps(record, sort_keys=True, default=str, separators=(",", ":"))
    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()


async def process_batch(
    records: list[dict[str, Any]],
) -> tuple[list[CostRecord], list[dict[str, Any]]]:
    valid_records: list[CostRecord] = []
    dead_letter: list[dict[str, Any]] = []
    for rec in records:
        try:
            valid_records.append(CostRecord.model_validate(rec))
        except ValidationError as exc:
            # Preserve the original payload intact; never mutate it in place.
            dead_letter.append(
                {
                    "payload": rec,
                    "validation_error": exc.errors(),
                    "hash": _fingerprint(rec),
                    "quarantined_at": datetime.now(PRODUCTION_TZ).isoformat(),
                }
            )
    return valid_records, dead_letter


async def run_async_ingestion_pipeline(source_urls: list[str], batch_size: int = 500) -> None:
    semaphore = asyncio.Semaphore(10)  # cap concurrent connections to prevent vendor rate-limiting
    connector = aiohttp.TCPConnector(limit=100, ttl_dns_cache=300)

    async with aiohttp.ClientSession(connector=connector) as session:
        tasks = [fetch_vendor_data(session, url, semaphore) for url in source_urls]
        raw_batches = await asyncio.gather(*tasks, return_exceptions=True)

        for batch in raw_batches:
            if isinstance(batch, BaseException):
                # One source failing must not cancel the others; route to alerting.
                continue

            # Process in chunks to bound peak memory during heavy month-end loads.
            for i in range(0, len(batch), batch_size):
                chunk = batch[i : i + batch_size]
                valid, failed = await process_batch(chunk)
                await route_to_ledger(valid)
                await route_to_dlq(failed)


# Placeholder routing functions for ERP / accounting-system integration.
async def route_to_ledger(records: list[CostRecord]) -> None:
    ...


async def route_to_dlq(records: list[dict[str, Any]]) -> None:
    ...

A note on the timestamp field and the quarantined_at stamp: both are timezone-aware. Cost data is reconciled against a studio’s operating day, and a bare UTC offset silently drifts across a daylight-saving boundary, so every temporal value is anchored to an IANA zone through zoneinfo rather than a naive datetime. When a shoot spans multiple locations, each unit’s records carry their own production zone, and the ledger normalizes on read — never by mutating the ingested value.

Guild and Contract Specifics: Encoding CBA Rules Into the Batch

Throughput is only useful if the records flowing through it are already compliant when they hit the ledger. Async batch processing is where collective bargaining agreement (CBA) enforcement is cheapest, because guild-rate validation runs as a parallel task alongside currency normalization instead of as a serial post-processing pass. Three agreements dominate the rules that a cost batch must respect. The International Alliance of Theatrical Stage Employees (IATSE) governs crew overtime, meal-penalty triggers, and turnaround; the Screen Actors Guild–American Federation of Television and Radio Artists (SAG-AFTRA) governs performer rates, overtime, and pension-and-health (P&H) contribution bases; and the Directors Guild of America (DGA) governs directorial-unit turnaround windows and overtime thresholds.

The practical pattern is to attach a rate table keyed by union_category, contract year, and jurisdiction, and to validate each payroll-derived CostRecord against it inside the same chunk pass that does schema validation. A rate table entry typically carries a base scale rate, an overtime multiplier (commonly 1.5x past an eight- or ten-hour threshold and 2x beyond a second threshold), a meal-penalty increment that accrues in fixed steps once a break is missed, and a fringe rate — the P&H and pension percentages layered on top of gross. Because those multipliers apply to money, every one of them is a Decimal operation; a fringe computed in floating point will disagree with the guild’s own figure by cents that compound into an audit finding.

When a payroll record references a category, year, or jurisdiction that the loaded rate table does not contain, the batch must not guess and must not stall. It routes that record through a controlled fallback path — the deterministic secondary-and-cached routing specified in Compliance Fallback Chains — so a missing rate table becomes an auditable exception rather than a halted close. The turnaround and overtime thresholds themselves are specified in DGA Overtime & Turnaround Rules, and the fringe-multiplier math that rides on every validated gross belongs to Pension & Health Fund Calculations. Encoding those thresholds directly into the validation step means a non-compliant payroll run is rejected into quarantine before it can reach the ledger — the whole subject of Guild Compliance & Rule Validation Automation — rather than being discovered weeks later during a guild audit. For international units, the per-transaction FX normalization that must precede any of this rate math is covered in depth by Async Batch Processing for Multi-Currency Shoots.

Error Handling and Quarantine: The Dead-Letter Queue

The dead-letter queue (DLQ) is what lets the pipeline guarantee zero data loss while still refusing to admit bad records. Its contract is strict: a rejected record is never dropped, never silently retried inline, and never mutated. Instead it is serialized exactly as received, stamped with a timezone-aware quarantined_at, annotated with the machine-readable ValidationError detail that explains why it failed, and fingerprinted with a SHA-256 hash of its canonical serialization. That fingerprint is the linchpin of the audit story: it lets even a malformed record be matched back to the original vendor submission without manual reconstruction, and it makes re-ingestion idempotent — a corrected record that reproduces the same canonical bytes reproduces the same hash, so a reconciliation tool can prove whether a fix was applied or a duplicate slipped in.

Two failure classes are handled differently on purpose. Record-level failures — a missing GL code, a mismatched department tag, a negative amount, an unknown union category — are validation outcomes and go to the DLQ with full context. Source-level failures — a timeout, a 503, a torn connection — are infrastructure outcomes; the orchestrator catches them via return_exceptions=True, routes an alert, and skips the source so its outage cannot poison the clean sources that ran alongside it. A line producer can then review the DLQ export in real time, correct department codes or supply a missing rate table, and re-ingest the affected records without restarting the entire batch. This quarantine discipline is the same boundary contract that Schema Validation & Error Handling enforces at the point of entry; async batch processing simply applies it under concurrency and at chunk scale.

Two failure classes routed apart while clean records continue to the ledger A batch orchestrator drives an async source fetch guarded by return_exceptions, which passes a raw batch to a validate-chunk gate; valid records commit to the append-only ledger. Two failure lanes peel off. A source-level failure — a 503, timeout, or torn connection — routes down from the source fetch to an alert-and-skip-source outcome while clean sources still commit. A record-level failure routes down from the validate gate to the dead-letter queue, which preserves the payload verbatim with its SHA-256 fingerprint, the ValidationError detail, and a timezone-aware quarantined_at stamp. Async batch run asyncio.gather() Source fetch async task · return_exceptions raw batch Validate chunk CostRecord.model_validate() valid Append-only ledger clean records commit on exception SOURCE-LEVEL FAILURE Alert + skip source 503 · timeout · torn connection clean sources still commit invalid record RECORD-LEVEL FAILURE Dead-letter queue payload verbatim · SHA-256 fingerprint ValidationError detail · quarantined_at (tz-aware)
Two failure classes, routed apart: a source-level exception skips the source and alerts while clean sources still commit; a record-level failure quarantines the payload with its hash, error detail, and timezone-aware stamp — and clean records continue to the ledger.

Verification: Confirming Ledger and Audit Output

An ingestion run is only trustworthy if its output can be checked without re-reading the source. Verification rests on three artifacts. First, the ledger entries: every accepted CostRecord should appear once in the append-only ledger, with its Decimal amount preserved to full precision, its currency and GL code intact, and its timezone-aware timestamp recorded against the correct production day. A reconciliation over the run should show that the count of ledger inserts plus the count of DLQ entries equals the count of records fetched — no record is ever both admitted and quarantined, and none simply vanishes.

Second, the audit log fields: each committed record and each quarantined record must carry, at minimum, its source URL, its ingestion pipeline execution identifier, its SHA-256 fingerprint, and a timezone-aware timestamp — written to append-only storage before the corresponding ledger transaction commits, so a crash mid-batch leaves a replayable record of intent rather than a gap. Because completion-bond lenders require explicit reconciliation of every dollar, that fingerprint is what lets a guarantor’s auditor tie any ledger line — or any rejected line — back to the exact vendor payload that produced it.

Third, the reconciliation report shape: a per-run summary that lists total fetched, total committed, total quarantined grouped by validation-error reason code, and total sources skipped with their failure class. When those figures agree with the guarantor’s expected daily cost movement, the run is provably complete. To confirm the concurrency layer itself is healthy in staging, enable asyncio.get_running_loop().set_debug(True) to surface any blocking call that exceeds the loop’s slow-callback threshold, as documented in the official Python asyncio documentation; the strict-typing and custom-validator behavior the schema depends on is detailed in the Pydantic documentation. Async batch processing, verified this way, turns cost ingestion from a reactive bottleneck into a proactive compliance engine: every vendor invoice, payroll run, and FX-adjusted transaction is processed with speed, precision, and a traceable path back to its origin.

Up one level: Cost Ingestion & Data Parsing Workflows.