Building a CAM Data Validation Layer

Field-level Pydantic models decide whether a single parsed invoice is well-formed, but a production CAM reconciliation still needs the layer that wraps those models: the code that runs every rule, decides accept-or-quarantine without raising, stamps each certified record with the ruleset that judged it, and streams thousands of invoices through the same gate during a month-end close. This page builds that orchestration — the quarantine writer, ruleset versioning, and streaming validation — as the deep implementation behind schema validation for parsed expense data, the validation stage of the broader Automated Invoice Parsing & Data Ingestion pipeline. Get the layer right and every downstream stage inherits a clean, penny-accurate, provenance-bound record; get it wrong and a single silently-coerced row seeds a tenant dispute six months later at year-end tie-out.

Context & When to Use This Approach #

Reach for a dedicated validation layer — not just inline model construction — the moment invoices stop arriving one at a time from a trusted source. The concrete triggers in a CRE feed are recognizable:

• A close window drops thousands of invoices at once from vendor portals, an email inbox, and an SFTP drop, and a single malformed record must never abort the batch that surrounds it.
• Accountants need a structured reason for every rejection — the exact rule, field, coordinates, and source hash — so they can triage a quarantine queue instead of re-reading raw PDFs.
• Your validation rules change over time (a property raises its capitalization threshold, a new exclusion is added), and an auditor must be able to reproduce exactly which ruleset certified any given line item months later.
• Records feed GL code mapping for CAM expenses and, ultimately, the pro rata allocation algorithms that distribute cost across tenants — so a defect that survives validation is a defect every tenant statement inherits.

If you are validating a handful of trusted invoices interactively, constructing a ParsedCAMInvoice and letting it raise is enough. The layer below is for the batch case, where raising is the wrong control flow: one bad record should be routed, not thrown, and the surrounding invoices must keep flowing.

This layer assumes the field-level models already exist. The ParsedCAMInvoice and CAMLineItem Pydantic models, the to_cents coercion helper, and the apply_lease_logic warning function are defined in schema validation for parsed expense data; this page imports them and builds the runtime around them.

Step-by-Step Implementation #

The pattern is a small state machine: an outcome type that can hold either a certified record or a structured error, a validator that converts exceptions into that outcome instead of propagating them, a ruleset stamp for reproducibility, a quarantine writer for rejects, and a generator that streams a whole batch through the gate with bounded memory. Every monetary value stays in Python’s decimal module — never float — so summing thousands of amounts never drifts an invoice out of the one-cent reconciliation tolerance.

Step 1 — Model the outcome as accept-or-quarantine, never a raw exception #

The whole point of the layer is that validation returns a value rather than throwing. Model that value explicitly: an accepted record carries the certified invoice and its ruleset version; a quarantined record carries a structured error payload an accountant can act on.

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
from typing import Optional, Union

# Field-level models live in the parent cluster's implementation.
from cam.schema import ParsedCAMInvoice, apply_lease_logic


@dataclass(frozen=True)
class ValidationError:
    """One machine-readable reason a record failed, for the review queue."""

    rule: str                       # e.g. "line_item_sum", "capitalization_threshold"
    field: Optional[str]            # offending field, when the rule is field-scoped
    message: str                    # human-readable triage text
    source_hash: str                # binds the reject to its exact source PDF


@dataclass(frozen=True)
class Accepted:
    invoice: ParsedCAMInvoice
    ruleset_version: str
    validated_at: datetime


@dataclass(frozen=True)
class Quarantined:
    invoice_id: str
    source_hash: str
    errors: list[ValidationError]
    validated_at: datetime


# A validation outcome is exactly one of these two shapes — never a half-valid row.
Outcome = Union[Accepted, Quarantined]

Step 2 — Version the ruleset so every certification is reproducible #

An accountant challenged on a two-year-old reconciliation must be able to answer “which rules certified this line?” Bind the answer to the record by stamping a ruleset version, and keep the tunable thresholds in a versioned config object rather than as literals scattered through the code.

@dataclass(frozen=True)
class Ruleset:
    """A versioned, per-property bundle of validation thresholds."""

    version: str
    capitalization_threshold: Decimal  # above this, a repair is cap-ex, not CAM
    min_ocr_confidence: Decimal        # below this, a field is too uncertain to trust
    tolerance: Decimal = Decimal("0.01")  # penny-level reconciliation band


# Registry: the exact ruleset that judged a record is recoverable by version.
RULESETS: dict[str, Ruleset] = {
    "v2": Ruleset(
        version="v2",
        capitalization_threshold=Decimal("5000.00"),
        min_ocr_confidence=Decimal("0.85"),
    ),
}

Step 3 — Run the layers and convert failures into an outcome #

This is the heart of the layer. Constructing ParsedCAMInvoice runs structural coercion and the line-item reconciliation identity; a Pydantic ValidationError becomes a Quarantined outcome rather than an exception. If the record is structurally sound, layer three’s lease logic runs, and any warnings quarantine it too. Only a record that clears both is Accepted.

from pydantic import ValidationError as PydanticValidationError


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


def validate_invoice(raw: dict, ruleset: Ruleset, source_hash: str) -> Outcome:
    """Validate one parsed invoice, returning an outcome instead of raising.

    Structural and financial rules run first (via the Pydantic model); lease
    logic runs only on a structurally sound record. Any failure routes the
    record to quarantine with a machine-readable payload.
    """
    invoice_id = str(raw.get("invoice_id", "UNKNOWN"))
    try:
        # Layers one and two: coercion + line-item reconciliation identity.
        invoice = ParsedCAMInvoice.model_validate(raw)
    except PydanticValidationError as exc:
        errors = [
            ValidationError(
                rule="schema",
                field=".".join(str(p) for p in err["loc"]) or None,
                message=err["msg"],
                source_hash=source_hash,
            )
            for err in exc.errors()
        ]
        return Quarantined(invoice_id, source_hash, errors, _now())

    # Layer three: lease logic that only the lease can adjudicate.
    warnings = apply_lease_logic(
        invoice,
        capitalization_threshold=ruleset.capitalization_threshold,
        min_confidence=ruleset.min_ocr_confidence,
    )
    if warnings:
        errors = [
            ValidationError(rule="lease_logic", field=None, message=w, source_hash=source_hash)
            for w in warnings
        ]
        return Quarantined(invoice.invoice_id, source_hash, errors, _now())

    return Accepted(invoice=invoice, ruleset_version=ruleset.version, validated_at=_now())

Step 4 — Persist rejects to a quarantine store an accountant can work #

A quarantined record that vanishes is worse than no validation at all. Serialize each reject as one JSON line — invoice id, source hash, timestamp, and every structured error — so the queue is both machine-queryable for metrics and human-readable for triage. In production this writer targets a database table or dead-letter queue; the shape of the payload is identical.

import json
from typing import TextIO


def write_quarantine(record: Quarantined, sink: TextIO) -> None:
    """Append one quarantined record to a JSONL store for review + metrics."""
    payload = {
        "invoice_id": record.invoice_id,
        "source_hash": record.source_hash,
        "validated_at": record.validated_at.isoformat(),
        "errors": [
            {"rule": e.rule, "field": e.field, "message": e.message}
            for e in record.errors
        ],
    }
    sink.write(json.dumps(payload) + "\n")

Step 5 — Stream a whole batch through the gate with bounded memory #

At portfolio scale the layer must process a batch without materializing it. A generator validates one record at a time, yields the accepted invoices for the next stage, and side-writes rejects to quarantine — so peak memory stays flat whether the batch holds ten invoices or ten thousand. This is the synchronous core that async batch processing for high-volume invoices later parallelizes across workers.

from typing import Iterable, Iterator


def validate_batch(
    records: Iterable[tuple[dict, str]],
    ruleset: Ruleset,
    quarantine_sink: TextIO,
) -> Iterator[Accepted]:
    """Stream (raw_record, source_hash) pairs through validation.

    Yields only Accepted records for the downstream mapper; every Quarantined
    record is written out and a running rejection count is kept for alerting.
    Memory stays bounded to a single record regardless of batch size.
    """
    seen = 0
    rejected = 0
    for raw, source_hash in records:
        seen += 1
        outcome = validate_invoice(raw, ruleset, source_hash)
        if isinstance(outcome, Quarantined):
            rejected += 1
            write_quarantine(outcome, quarantine_sink)
            continue
        yield outcome

    # A spiking rejection rate for one vendor signals a changed template —
    # far cheaper to catch here than in a failed year-end tie-out.
    if seen and rejected / seen > 0.10:
        # Hook a real alert here (metrics emit, pager, dashboard annotation).
        pass

Gotchas & Known Limitations #

The failure modes that actually bite are in the orchestration, not the field rules. Treat this as a pre-flight checklist before trusting the layer against a new feed:

• Raising instead of routing. If any rule escapes validate_invoice as an exception, one malformed invoice aborts the whole batch. Catch PydanticValidationError explicitly and convert it; never let the model raise past the layer boundary.
• Unstamped accepted records. An Accepted record without a ruleset_version is un-auditable — you cannot later prove which thresholds judged it. Stamp the version at acceptance time, and never mutate a RULESETS entry in place; publish a new version instead.
• Silent coercion masquerading as success. If a rule “fixes” a value rather than rejecting it, the layer’s accept-or-quarantine contract is broken. Coercion is confined to canonicalizing money (to_cents) and dates; anything the lease must adjudicate is flagged, not repaired.
• Losing provenance on the reject path. A quarantined record without its source_hash cannot be traced back to the exact PDF, so an accountant re-hunts the source by hand. Bind the hash on both the accept and reject branches — it comes from the stored, immutable source document, not from the parse.
• Credits and reversals treated as malformed. Vendor credit memos arrive as negative amounts and trip a gt=0 field constraint, landing legitimate credits in quarantine. Decide the sign contract upstream at extraction, not by loosening a validator.
• float anywhere in the amount path. A single float(cell) upstream reintroduces binary drift that a thousand summed rows amplify past the one-cent tolerance, producing false rejects on correct invoices. Keep the entire path in Decimal.
• Materializing the batch. Building a list of every parsed record before validating defeats the streaming design and risks an out-of-memory kill during close. Feed validate_batch an iterator, not a list.

Verification #

Confirm the layer’s contract by asserting on both branches: a clean invoice returns Accepted with the right ruleset stamp, and each defect class returns Quarantined with the specific rule that should have fired — not merely “some error.” Assert on payload contents so a future refactor cannot quietly change which rule caught a defect.

import io
from decimal import Decimal

RS = RULESETS["v2"]


def _clean_raw() -> dict:
    return {
        "invoice_id": "INV-001",
        "vendor_name": "Acme Facilities",
        "invoice_date": "2026-01-15",
        "property_id": "PROP-42",
        "total_amount": "1075.50",
        "line_items": [
            {"description": "HVAC", "gl_code": "5100-HVAC", "amount": "900.00", "tax_amount": "75.50"},
            {"description": "Sweep", "gl_code": "5200-LOT", "amount": "100.00"},
        ],
    }


def test_clean_invoice_is_accepted_and_stamped() -> None:
    outcome = validate_invoice(_clean_raw(), RS, source_hash="sha256:abc")
    assert isinstance(outcome, Accepted)
    assert outcome.ruleset_version == "v2"
    assert outcome.invoice.total_amount == Decimal("1075.50")


def test_transposed_total_quarantines_on_schema_rule() -> None:
    raw = _clean_raw()
    raw["total_amount"] = "1750.50"  # digits transposed from 1075.50
    outcome = validate_invoice(raw, RS, source_hash="sha256:def")
    assert isinstance(outcome, Quarantined)
    assert any(e.rule == "schema" for e in outcome.errors)
    assert outcome.source_hash == "sha256:def"  # provenance preserved on reject


def test_large_repair_quarantines_on_lease_logic() -> None:
    raw = _clean_raw()
    raw["line_items"] = [{"description": "Roof", "gl_code": "5300-ROOF",
                          "amount": "12000.00", "tax_amount": "0.00"}]
    raw["total_amount"] = "12000.00"
    outcome = validate_invoice(raw, RS, source_hash="sha256:ghi")
    assert isinstance(outcome, Quarantined)
    assert any("cap-ex" in e.message for e in outcome.errors if e.rule == "lease_logic")


def test_batch_routes_accepts_out_and_rejects_to_sink() -> None:
    sink = io.StringIO()
    good = (_clean_raw(), "sha256:1")
    bad_raw = _clean_raw()
    bad_raw["total_amount"] = "9999.99"
    bad = (bad_raw, "sha256:2")
    accepted = list(validate_batch([good, bad], RS, sink))
    assert len(accepted) == 1                      # only the clean record flows on
    assert "sha256:2" in sink.getvalue()           # the reject was persisted
    assert '"rule": "schema"' in sink.getvalue()   # with its structured reason

The batch test is the one that matters most: it proves the layer routes rather than raises — the clean record reaches the downstream mapper while the broken one is persisted with a machine-readable reason and never blocks the invoice beside it. Extend it with property-based tests that generate random line-item sets, compute the true total in Decimal, and assert the reconciling invoice always accepts while a total perturbed by more than a cent always quarantines on the schema rule.

Once the layer is green, its accepted output flows on to GL code mapping for CAM expenses with a guarantee the mapper can rely on, and the whole design belongs to schema validation for parsed expense data; if your rejections cluster around one vendor’s scanned statements, the fix usually lives upstream in handling multi-page commercial invoices in Python.

Related #

• Schema validation for parsed expense data — the parent stage that defines the Pydantic models and layered rules this layer orchestrates.
• GL code mapping for CAM expenses — the downstream stage that consumes every record this layer certifies.
• Async batch processing for high-volume invoices — parallelizes the streaming validator across workers for month-end close volume.