Core Architecture & Compliance Mapping

JSON Schema for Municipal Disposal Tracking

Draft-07 schemas with strict type bounds, fallback chains, and audit-ready logging.

You need a single Draft-07 JSON Schema that validates every municipal disposal-tracking payload — GPS pings, scalehouse weights, driver Hours-of-Service metadata, and fallback endpoints — and rejects anything malformed at the ingestion edge before it corrupts a route or an audit trail.

This page delivers the schema, a precompiled jsonschema validator, and the exact structured audit record it emits. It is the Draft-07, wire-format counterpart to the strict Pydantic contract described in the parent Route Schema Design reference: use Pydantic when you own the producer in Python, and use this JSON Schema when the payload arrives over HTTP from telematics gateways or third-party scalehouse vendors that only speak JSON. Both enforce the same invariant the Core Architecture & Compliance Mapping framework demands — identical input must always produce an identical audit hash, so no field may be silently coerced or defaulted.

Environment & Data Prerequisites

Target Python 3.10+. Pin the validator and its date backport so schema behavior is byte-reproducible across ingestion nodes:

python -m pip install \
  jsonschema==4.22.0 \
  rfc3339-validator==0.1.4 \
  python-dateutil==2.9.0.post0

The rfc3339-validator package activates the built-in date-time format check in jsonschema; without it, format: "date-time" is silently ignored and naive timestamps slip through. A single disposal-tracking payload has this shape — every field carries a defined unit and range:

Field Type / format Unit Expected range
schema_version string const — exactly "1.2.0"
route_id string, ^RT-[A-Z0-9]{6}$ — e.g. RT-8A4F2C
timestamp_utc string, RFC 3339 UTC tz-aware, Z or ±hh:mm
gps_telemetry.lat / .lon number WGS84 decimal degrees municipal bounding box
compliance_block.vehicle_gvw integer pounds ≥ 10000
compliance_block.hazmat_flag boolean — strict true / false
weight_payload.*_lbs number, multipleOf 0.01 pounds ≥ 0
fallback_chain[] array of objects — ≥ 1, unique priority

Coordinate bounds are the last line of defense against GPS drift; raw fixes should already have passed the kinematic gate in Validating GPS Coordinates in Python before they are promoted into a disposal-tracking payload. The weight_payload figures originate at the scalehouse and feed the same gross-weight ceiling the solver binds as a cumulative dimension under Capacity & Weight Limits.

Draft-07 validation gate: one payload, five typed sub-checks, two audited outcomes A left-to-right pipeline. A JSON disposal-tracking payload flows into a validation-gate box holding five stacked sub-checks — route/version identity, GPS bounding box, DOT compliance fields, weight multipleOf, and a dashed application-layer unique-priority check. A junction labelled "valid?" splits into a green accepted branch (audit hash stamped, sent to the dispatch queue) and a red rejected branch (field-level error, sent to the dead-letter queue). INGESTION EDGE disposal payload application/json HTTP · telematics + scalehouse Draft-07 Validation Gate additionalProperties: false route_id · schema_version pattern ^RT-…$ · const "1.2.0" gps_telemetry lat/lon within municipal bounding box compliance_block vehicle_gvw ≥ 10000 · hazmat_flag bool weight_payload multipleOf 0.01 · minimum 0 fallback_chain unique priority · app-layer cross-check FIRST FAILING CHECK STOPS THE PAYLOAD iter_errors() valid? true PASS audit hash stamped byte-reproducible record · status PASS → dispatch / routing queue false FAIL field-level error { field, message, validator } → dead-letter queue
One Draft-07 gate, five typed sub-checks, two audited outcomes: every payload is either hash-stamped for dispatch or rejected with a granular field-level error to the dead-letter queue.

Step-by-Step Implementation

The build has four steps: declare the precompiled schema, add a strict ISO-8601 format check, wrap validation with a cross-item priority-uniqueness rule the pure schema cannot express, then run a payload through the gate and serialize the result.

Step 1 — Declare the precompiled Draft-07 schema

Root and every nested object set additionalProperties: false so an unexpected key is a rejection, not a shrug. Numeric bounds encode municipal reality: vehicle_gvw starts at 10 000 lb (a Class 3+ truck), weight_payload uses multipleOf: 0.01 to pin scalehouse readings to whole cents of a pound and eliminate floating-point ambiguity during reconciliation.

DISPOSAL_SCHEMA = {
    "$schema": "http://json-schema.org/draft-07/schema#",
    "type": "object",
    "required": [
        "schema_version", "route_id", "timestamp_utc",
        "compliance_block", "weight_payload", "fallback_chain",
    ],
    "properties": {
        "schema_version": {"type": "string", "const": "1.2.0"},
        "route_id": {"type": "string", "pattern": "^RT-[A-Z0-9]{6}$"},
        "timestamp_utc": {"type": "string", "format": "date-time"},
        "gps_telemetry": {
            "type": "object",
            "required": ["lat", "lon"],
            "properties": {
                "lat": {"type": "number", "minimum": 32.0, "maximum": 34.5},
                "lon": {"type": "number", "minimum": -119.0, "maximum": -116.5},
            },
            "additionalProperties": False,
        },
        "compliance_block": {
            "type": "object",
            "required": ["driver_license_state", "vehicle_gvw", "hazmat_flag"],
            "properties": {
                "driver_license_state": {"type": "string", "pattern": "^[A-Z]{2}$"},
                "vehicle_gvw": {"type": "integer", "minimum": 10000},
                "hazmat_flag": {"type": "boolean"},
            },
            "additionalProperties": False,
        },
        "weight_payload": {
            "type": "object",
            "required": ["gross_weight_lbs", "tare_weight_lbs"],
            "properties": {
                "gross_weight_lbs": {"type": "number", "multipleOf": 0.01, "minimum": 0},
                "tare_weight_lbs": {"type": "number", "multipleOf": 0.01, "minimum": 0},
            },
            "additionalProperties": False,
        },
        "fallback_chain": {
            "type": "array",
            "minItems": 1,
            "items": {
                "type": "object",
                "required": ["priority", "endpoint_uri", "timeout_ms"],
                "properties": {
                    "priority": {"type": "integer", "minimum": 1},
                    "endpoint_uri": {"type": "string", "format": "uri"},
                    "timeout_ms": {"type": "integer", "minimum": 500, "maximum": 5000},
                },
                "additionalProperties": False,
            },
        },
    },
    "additionalProperties": False,
}

Step 2 — Enforce strict ISO-8601 and precompile the validator

The default date-time check accepts values a naive parser would mangle later. A tightened regex requires an explicit UTC offset (Z or ±hh:mm), which is the single most common source of daylight-saving route bugs. Precompiling one Draft7Validator instance avoids re-parsing the schema AST on every packet during peak collection hours.

import re
from typing import Any
from jsonschema import Draft7Validator, FormatChecker

ISO8601 = re.compile(
    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(\.\d+)?([+-]\d{2}:\d{2}|Z)$"
)

format_checker = FormatChecker()


@format_checker.checks("date-time")
def _is_strict_iso8601(instance: Any) -> bool:
    return isinstance(instance, str) and bool(ISO8601.match(instance))


VALIDATOR = Draft7Validator(DISPOSAL_SCHEMA, format_checker=format_checker)

Step 3 — Bind the cross-item constraint the schema cannot express

Draft-07 can require a non-empty fallback_chain, but it cannot assert that every priority in the array is distinct. Duplicate priorities make degradation order ambiguous during a cellular dead zone, so the application layer enforces uniqueness after the structural pass. The JSONFormatter below is the custom formatter used site-wide — the standard logging module ships no JSON formatter, so structured audit output requires one.

import json
import logging
from typing import Any


class JSONFormatter(logging.Formatter):
    """Custom JSON formatter — the standard logging module has no built-in JSONFormatter."""

    _BASE = set(logging.LogRecord("", 0, "", 0, "", (), None).__dict__)

    def format(self, record: logging.LogRecord) -> str:
        log_obj = {
            "timestamp": self.formatTime(record),
            "level": record.levelname,
            "message": record.getMessage(),
        }
        log_obj.update({k: v for k, v in record.__dict__.items() if k not in self._BASE})
        return json.dumps(log_obj, separators=(",", ":"))


logger = logging.getLogger("disposal.ingestion")
_handler = logging.StreamHandler()
_handler.setFormatter(JSONFormatter())
logger.addHandler(_handler)
logger.setLevel(logging.INFO)


def validate_disposal_payload(payload: dict[str, Any]) -> dict[str, Any]:
    """Stateless validation: structural schema pass + unique-priority cross-check."""
    errors: list[dict[str, str]] = []

    for err in VALIDATOR.iter_errors(payload):
        errors.append({
            "field": ".".join(map(str, err.absolute_path)) or "<root>",
            "message": err.message,
            "validator": err.validator,
        })

    chain = payload.get("fallback_chain")
    if isinstance(chain, list):
        priorities = [item.get("priority") for item in chain if isinstance(item, dict)]
        if len(priorities) != len(set(priorities)):
            errors.append({
                "field": "fallback_chain",
                "message": "Duplicate priority values detected in fallback_chain",
                "validator": "unique_priority",
            })

    return {"valid": not errors, "errors": errors}

Step 4 — Run the gate and serialize the outcome

One entry point validates the payload, emits a structured audit record, and branches to dispatch or to a dead-letter queue. Nothing invalid ever reaches the solver; recovery is an explicit, logged decision made downstream in the Fallback Routing Logic layer.

from datetime import datetime, timezone

MOCK_PAYLOAD = {
    "schema_version": "1.2.0",
    "route_id": "RT-8A4F2C",
    "timestamp_utc": "2024-03-15T08:42:11Z",
    "gps_telemetry": {"lat": 33.4484, "lon": -118.3215},
    "compliance_block": {
        "driver_license_state": "CA",
        "vehicle_gvw": 26000,
        "hazmat_flag": False,
    },
    "weight_payload": {"gross_weight_lbs": 32750.50, "tare_weight_lbs": 18500.00},
    "fallback_chain": [
        {"priority": 1, "endpoint_uri": "https://telemetry.muni.gov/v2/ingest", "timeout_ms": 2000},
        {"priority": 2, "endpoint_uri": "https://backup.muni.gov/v2/ingest", "timeout_ms": 3000},
    ],
}


def process_ingestion(payload: dict[str, Any]) -> bool:
    result = validate_disposal_payload(payload)
    logger.info(
        "payload_validation",
        extra={
            "route_id": payload.get("route_id", "unknown"),
            "validation_status": "PASS" if result["valid"] else "FAIL",
            "audited_at": datetime.now(timezone.utc).isoformat(),
            "errors": result["errors"] or None,
        },
    )
    if result["valid"]:
        return True   # enqueue for routing / dispatch
    return False      # route to dead-letter queue for ops review


if __name__ == "__main__":
    process_ingestion(MOCK_PAYLOAD)

Compliance Output

An accepted payload emits exactly one newline-delimited JSON audit record. Each field exists for a specific regulatory reason, so an auditor can trace a dispatched route back to the authority that governs it:

{"timestamp":"2024-03-15 08:42:15,112","level":"INFO","message":"payload_validation","route_id":"RT-8A4F2C","validation_status":"PASS","audited_at":"2024-03-15T08:42:15.112000+00:00","errors":null}
  • route_id — the immutable key that joins this record to the manifest and the solver’s audit hash; without it the trail cannot be reconstructed.
  • validation_status — proves the payload was gated, not assumed; FAIL records carry the offending field, which is the evidence an auditor reviews.
  • audited_at — a tz-aware ingestion timestamp distinct from timestamp_utc, establishing when the record entered the compliance system.
  • compliance_block.vehicle_gvw maps to Gross Vehicle Weight Rating under 49 CFR 571 and the posted axle and bridge limits under 23 CFR 658; the schema enforces the gross floor before the solver binds per-axle limits.
  • compliance_block.hazmat_flag = true triggers the EPA hazardous-waste manifest requirement under 40 CFR 262 Subpart B, whose e-Manifest is EPA Form 8700-22, and the hazardous-routing restrictions of 49 CFR 397.
  • compliance_block.driver_license_state anchors the driver record used to check driving-time ceilings under Hours of Service, 49 CFR 395 — the field-by-field translation of those ceilings into solver limits lives in the DOT/FMCSA Rule Mapping reference.

Because null is prohibited on audit-critical paths and every weight is pinned to multipleOf: 0.01, scalehouse reconciliation is exact and the record is byte-reproducible for a DOT audit. Access controls and retention around these manifest records are covered in Security & Access Boundaries.

Verification

One pytest module confirms the gate accepts a well-formed payload and rejects each failure class it exists to catch. Run it with pytest test_disposal_schema.py -q.

import copy
import pytest

BASE = MOCK_PAYLOAD  # reuse the valid fixture from Step 4


def test_valid_payload_passes():
    assert validate_disposal_payload(BASE)["valid"] is True


def test_naive_timestamp_is_rejected():
    bad = copy.deepcopy(BASE)
    bad["timestamp_utc"] = "2024-03-15T08:42:11"  # no UTC offset
    result = validate_disposal_payload(bad)
    assert result["valid"] is False
    assert any(e["field"] == "timestamp_utc" for e in result["errors"])


def test_underweight_gvw_is_rejected():
    bad = copy.deepcopy(BASE)
    bad["compliance_block"]["vehicle_gvw"] = 8000  # below 10000 lb floor
    assert validate_disposal_payload(bad)["valid"] is False


def test_duplicate_fallback_priority_is_rejected():
    bad = copy.deepcopy(BASE)
    bad["fallback_chain"][1]["priority"] = 1  # collides with entry 0
    result = validate_disposal_payload(bad)
    assert any(e["validator"] == "unique_priority" for e in result["errors"])


def test_unknown_field_is_rejected():
    bad = copy.deepcopy(BASE)
    bad["compliance_block"]["driver_note"] = "left gate open"
    assert validate_disposal_payload(bad)["valid"] is False

In production, alert on the payload_validation event grouped by field when validation_status is FAIL: a spike on one field signals an upstream producer that has drifted from the contract.

Common Errors

jsonschema.exceptions.SchemaError: 'date-time' is not a valid format — you passed format_checker but the format libraries are missing. Root cause: rfc3339-validator is not installed, so FormatChecker cannot register date-time. Fix: install the pinned prerequisites above; the custom @format_checker.checks("date-time") in Step 2 also guarantees the check runs even if the optional backend is absent.

A naive timestamp passes validation unexpectedly — you constructed Draft7Validator(DISPOSAL_SCHEMA) without the format_checker argument. Root cause: format keywords are annotations, not assertions, in JSON Schema; they are inert unless a checker is attached. Fix: always pass format_checker=format_checker, as in Step 2, so format: "date-time" and format: "uri" actually fire.

TypeError: unhashable type: 'dict' from the priority check — a fallback_chain item is malformed (a nested list or missing priority). Root cause: the uniqueness cross-check ran on data the structural pass would also reject. Fix: the guard in Step 3 only reads priority from dict items, so let VALIDATOR.iter_errors report the structural failure first; do not reorder the cross-check ahead of the schema pass.

Up: Route Schema Design