Schema Validation Rules

In construction project tracking, the reliability of automated change-order workflows depends entirely on how rigorously extracted document data conforms to predefined structural expectations. The specific sub-problem this page solves is how to turn noisy extracted fields into a typed, contract-aligned record — and how to reject everything that does not fit before it can touch a cost ledger or a schedule. When Automated Document Ingestion & Parsing pipelines process subcontractor submissions, architect directives, and owner change logs, raw text and tabular data must be normalized into a deterministic contract before it can trigger cost rollups or schedule adjustments. Schema validation is the gatekeeper between unstructured field data and actionable project intelligence. Without it, estimators receive misallocated cost codes, project managers approve unverified schedule deltas, and the audit trail develops gaps that surface months later during a payment dispute.

This page is the schema subsystem of the ingestion pipeline. It assumes documents have already been captured and the raw fields lifted out of them; its job is to define the canonical change-order model, coerce extraction artifacts into it, and hand a clean, typed payload to downstream routing. It is intentionally narrow: extraction lives upstream in Field Extraction Techniques, and failure recovery lives downstream in Error Handling Protocols. This page owns the contract in between.

Prerequisites

The validation layer is a thin, dependency-light module by design — the contract has to be importable from every other stage without dragging in parsing libraries. Before implementing the patterns below you need:

• Python 3.11+ with pydantic v2 for typed validation. The model_config, field_validator(..., mode="before"), and Literal/Annotated constraint features used here are v2-specific.
• The standard decimal, enum, re, and logging modules. All financial fields use Decimal, never float, to avoid floating-point drift in cost aggregations.
• Extracted, but not yet typed, payloads. The input to this layer is a dict of raw strings produced by field extraction techniques and the PDF/Excel sync pipelines that feed them. Values like "+$14,250.00" and "+14 days" are expected; the schema coerces them.
• A confidence score per field. Validation does not invent confidence — it is supplied by the extraction stage and threaded through so routing can use the site-canonical bands: 0.92 and above auto-routes, 0.75–0.92 goes to human review, and below 0.75 is quarantined.
• A taxonomy binding. Each document should already carry a cost account so the validated record can name where it belongs. Cost codes follow CSI MasterFormat XX XX XX divisions and the project work breakdown structure element pattern; the schema enforces both as regex-validated fields rather than free strings.

Contract-Aligned Schema Architecture

Production-grade validation begins with schema design that mirrors contractual and accounting requirements rather than a generic data model. A change-order schema must enforce strict data types, mandatory fields, and enumerated values that reflect actual contract-administration constraints. The canonical record carries the following fields:

By anchoring the schema to real contract workflows, the pipeline eliminates ambiguous data states before they reach the project-tracking dashboard. Every record carries the minimum context required for downstream processing, so financial models and critical-path schedules consume predictable, type-safe payloads. The discipline and cost-code constraints in particular do double duty: they validate structure and assert that the change has been mapped to a real account, which is what makes deterministic budget code standardization possible further down the line.

Integration with Extraction Pipelines

Raw construction documents rarely arrive in a uniform format, which makes the transition from extraction to validation the most fragile stage of the pipeline. Subcontractors submit PDFs with embedded tables, owners circulate Excel change logs with merged cells, and field superintendents capture handwritten directives that require optical recognition. As data moves through the PDF/Excel sync pipelines, the parsing layer maps heterogeneous cell structures and text blocks toward the canonical schema, and any scanned source first passes through OCR preprocessing.

The schema’s job is to absorb that heterogeneity deterministically. When an extraction engine returns "+$14,250.00" or "+14 days", the validation layer coerces these strings into validated Decimal and int types while preserving an audit trail of the original text. Cross-validation between extracted line items and master contract totals catches silent overruns. Critically, validation failures must never halt the entire batch: compliant records proceed while non-compliant ones route to isolated error queues for manual review, a contract handed off to the error handling protocols.

Step-by-Step Implementation

The following builds a production-ready validation layer with Pydantic v2. It enforces type safety, construction-specific business logic, and structured error reporting suitable for high-throughput automation.

Step 1: Pin the controlled vocabularies as typed enums

Approval states and disciplines are not free strings — they are state machines and routing keys. Encoding them as Enum and Literal types means an out-of-vocabulary value fails at the schema boundary rather than corrupting a downstream branch.

import logging
import re
from decimal import Decimal, InvalidOperation
from enum import Enum
from typing import Literal, Optional
from pydantic import BaseModel, Field, field_validator, ValidationError

# Structured logging keeps the validation stage observable in production.
logger = logging.getLogger("schema_validation")

class ApprovalState(str, Enum):
    PENDING = "PENDING"
    UNDER_REVIEW = "UNDER_REVIEW"
    APPROVED = "APPROVED"
    REJECTED = "REJECTED"
    EXECUTED = "EXECUTED"

# Discipline codes are site-canonical and drive reviewer routing.
Discipline = Literal["ARCH", "STR", "MEP", "CIV", "ELEC", "PLMB"]

Step 2: Declare the canonical model with strict field constraints

strict=True blocks silent type coercion (a string "5" will not become an int), and extra="forbid" rejects any field the contract does not define — a hallucinated OCR column cannot smuggle itself into the ledger.

class ChangeOrderSchema(BaseModel):
    model_config = {"strict": True, "extra": "forbid"}

    change_order_id: str = Field(pattern=r"^CO-\d{4}-\d{3,}$")
    contract_ref: str = Field(min_length=3, max_length=20)
    cost_code: str = Field(pattern=r"^\d{2} \d{2} \d{2}$")          # MasterFormat XX XX XX
    wbs_element: str = Field(
        pattern=r"^PROJ-\d{3}-(ARCH|STR|MEP|CIV|ELEC|PLMB)-\d{2}$"  # WBS PROJ-NNN-DIV-NN
    )
    discipline: Discipline
    cost_impact: Decimal = Field(ge=0, decimal_places=2)
    schedule_impact_days: int = Field(ge=0)
    approval_state: ApprovalState
    field_confidence: Decimal = Field(ge=0, le=1)
    extracted_raw_cost: Optional[str] = None
    extracted_raw_days: Optional[str] = None

Step 3: Coerce extraction artifacts with mode="before" validators

Extraction returns human-formatted strings; the contract wants typed values. Before-validators run prior to type enforcement so currency symbols, thousands separators, and "days" suffixes are stripped without weakening the declared types. They never default a missing financial value — an unparseable cost raises rather than becoming zero, because a fabricated zero in a cost ledger is far more damaging than a rejected record.

    @field_validator("cost_impact", mode="before")
    @classmethod
    def parse_currency(cls, v: object) -> Decimal:
        if isinstance(v, Decimal):
            return v
        if not isinstance(v, str):
            raise ValueError("Cost impact must be a string or Decimal")
        cleaned = v.replace("$", "").replace(",", "").strip().lstrip("+")
        try:
            val = Decimal(cleaned)
        except InvalidOperation as exc:
            raise ValueError(f"Invalid currency format: {v}") from exc
        if val < 0:
            raise ValueError("Contractually prohibited negative cost impact")
        return val

    @field_validator("schedule_impact_days", mode="before")
    @classmethod
    def parse_schedule_delta(cls, v: object) -> int:
        if isinstance(v, int):
            return v
        if not isinstance(v, str):
            raise ValueError("Schedule impact must be a string or int")
        cleaned = re.sub(r"(days|day|d)", "", v, flags=re.IGNORECASE).strip().lstrip("+")
        try:
            return int(cleaned)
        except ValueError as exc:
            raise ValueError(f"Invalid schedule format: {v}") from exc

Step 4: Validate and route by confidence band

The entry point returns a typed model on success and surfaces structured diagnostics on failure. Routing keys off the confidence band carried on the record — the same 0.92 / 0.75 thresholds used everywhere in the pipeline — so a structurally valid but low-confidence extraction is held for human review rather than auto-posted.

def validate_change_order_record(raw_payload: dict) -> ChangeOrderSchema:
    """Validate one extracted change-order record against the production schema.

    Returns a validated model or raises ValidationError with structured diagnostics.
    """
    validated = ChangeOrderSchema(**raw_payload)  # raises ValidationError on any breach
    logger.info("Record %s validated", validated.change_order_id)
    return validated

def route_validated_record(record: ChangeOrderSchema) -> str:
    """Map a validated record to a pipeline disposition by confidence band."""
    if record.field_confidence >= Decimal("0.92"):
        return "AUTO_ROUTE"          # post straight to the ledger
    if record.field_confidence >= Decimal("0.75"):
        return "HUMAN_REVIEW"        # queue for an estimator
    return "QUARANTINE"              # hold in the dead-letter queue

This layer enforces workflow boundaries by rejecting extraneous fields (extra="forbid"), normalizing extraction artifacts via before-validators, and surfacing precise error traces that the error handling protocols translate into SCHEMA_* codes. The extracted_raw_* fields preserve the original OCR/parse output for audit compliance without polluting the canonical model. In high-volume environments — bid-period bursts, end-of-month change-log dumps — validation should execute inside the async batching workflows so latency spikes stay isolated and never block a request thread.

Schema and Configuration Reference

The validators above depend on a handful of tunable values. Keeping them in one configuration block — rather than scattered as magic numbers — lets contract administrators adjust behavior per project without touching the model.

Validation rules summarized: identifiers and codes are regex-gated; financial fields are non-negative Decimal with fixed precision; schedule deltas are non-negative integers; vocabularies are closed enums/literals; and confidence is a bounded Decimal used only for routing, never for accept/reject.

Verification and Testing

Schema code is only trustworthy if its boundaries are tested with both clean and adversarial inputs. The following assertions confirm the contract accepts a well-formed record and rejects each failure class deterministically.

import pytest
from decimal import Decimal

GOOD = {
    "change_order_id": "CO-2024-089",
    "contract_ref": "MA-7781",
    "cost_code": "03 30 00",                 # Cast-in-Place Concrete
    "wbs_element": "PROJ-114-STR-02",
    "discipline": "STR",
    "cost_impact": "+$14,250.00",
    "schedule_impact_days": "+14 days",
    "approval_state": "PENDING",
    "field_confidence": Decimal("0.95"),
}

def test_accepts_and_coerces_clean_record():
    rec = validate_change_order_record(GOOD)
    assert rec.cost_impact == Decimal("14250.00")     # currency string coerced
    assert rec.schedule_impact_days == 14             # "+14 days" coerced
    assert route_validated_record(rec) == "AUTO_ROUTE"

def test_rejects_malformed_cost_code():
    bad = {**GOOD, "cost_code": "033000"}             # missing the XX XX XX spacing
    with pytest.raises(ValidationError):
        validate_change_order_record(bad)

def test_rejects_unknown_extra_field():
    bad = {**GOOD, "ghost_column": "hallucinated"}    # extra="forbid"
    with pytest.raises(ValidationError):
        validate_change_order_record(bad)

def test_low_confidence_record_is_held():
    rec = validate_change_order_record({**GOOD, "field_confidence": Decimal("0.80")})
    assert route_validated_record(rec) == "HUMAN_REVIEW"

For a quick manual check, python -c "from validation import validate_change_order_record; ..." against a single payload prints either the model’s model_dump_json() or the ValidationError with the exact offending field and constraint — enough to confirm wiring before pointing the stage at a live queue.

Troubleshooting

1. European decimal formats fail parse_currency. A value like "14.250,00" (period thousands, comma decimal) parses to the wrong number or raises InvalidOperation. Root cause: the cleaner assumes the US ,/. convention. Fix by detecting locale at the extraction stage and normalizing to a canonical .-decimal string before validation, rather than guessing inside the validator.
2. Stamped-drawing change orders quarantine on confidence, not structure. A perfectly structured record still routes to QUARANTINE because field_confidence came in below 0.75 from OCR on a stamped sheet. This is correct behavior, not a schema bug — raise the source quality in OCR preprocessing; do not lower the threshold to force a post.
3. extra="forbid" rejects records after a real contract change. When a project legitimately adds a field, every payload fails until the model is updated. Version the schema (schema_version in the payload) and gate the model so old and new revisions validate against their matching contract, instead of loosening forbid.
4. Negative cost_impact raises on a valid credit change order. Acceleration and credit COs are real. Root cause: ge=0 plus the negative guard. Fix by flipping cost_impact.allow_negative for the affected contract and relaxing the constraint conditionally — do not strip the sign, which would silently flip a credit into a charge.
5. strict=True rejects a numeric cost that arrives as an int. Strict mode will not coerce 14250 (int) into Decimal. Because the before-validator only handles str/Decimal, send financial values as strings from extraction, or extend parse_currency to accept int explicitly — never disable strict mode globally to paper over one field.

Frequently Asked Questions

Related

• Validating extracted RFI fields against custom JSON schemas
• Field Extraction Techniques
• Error Handling Protocols
• Async Batching Workflows
• WBS Mapping Strategies

For external interoperability, adherence to the JSON Schema specification keeps these contracts portable to ERP and accounting systems, and the Pydantic documentation details the v2 validator behavior the implementation relies on.

← Back to Automated Document Ingestion & Parsing