Validating JSON billing payloads with Pydantic

This page walks through the exact Python needed to turn raw, untrusted JSON billing exports — from Cost and Usage Reports, Azure Cost Management, or a third-party cost API — into deterministic, Decimal-safe, typed cost records before a single row reaches your ledger.

Back to: Schema Validation for Billing Data

Malformed JSON, missing dimensional tags, a cost silently serialized as an IEEE-754 float, or a timestamp arriving in a mixed offset are all corruption vectors that surface only at month-end reconciliation, long after the bad record has been summed into a chargeback total. The fix is to make the ingestion boundary the contract layer that turns unpredictable billing exports into type-safe cost records: a strict Pydantic v2 model that coerces money to Decimal, canonicalizes time to UTC, recomputes the cost math, and routes anything that fails to a dead-letter queue instead of into the aggregation layer of your Metric Extraction & Aggregation Pipelines. This guide builds that validator end to end, then wires it into the same fail-fast discipline used across error handling in cost pipelines.

Prerequisites

Confirm the following are in place before running the validator.

  • Source access (IAM): validation runs on JSON you have already exported, but the fetch step that feeds it needs read-only access to wherever the billing export lands. For AWS Cost and Usage Reports delivered to S3, scope a least-privilege policy to the export bucket only — never reuse an admin credential for read ingestion, in line with broader access control for cost data.

    {
  "Version": "2012-10-17",
  "Statement": [
    {
      "Sid": "ReadBillingExports",
      "Effect": "Allow",
      "Action": ["s3:GetObject", "s3:ListBucket"],
      "Resource": [
        "arn:aws:s3:::org-cur-exports",
        "arn:aws:s3:::org-cur-exports/*"
      ]
    }
  ]
}

  • Python: 3.10 or newer (the model uses typing.Annotated and X | Y unions).

  • Libraries: install Pydantic v2. No cloud SDK is required for the validation layer itself — that isolation is deliberate, so the same model can validate payloads from any provider.

    pip install "pydantic>=2.6"

Step-by-Step Implementation

The validator defines a strict record contract, coerces every monetary field through Decimal before Pydantic’s type checker runs, canonicalizes timestamps to timezone-aware UTC, reconciles the reported total against the recomputed one, and wraps the whole thing in an ingest function that routes failures to a dead-letter queue. The canonical total a validated record must satisfy is derived from its components:

$$\text{total_cost} = \text{usage_quantity} \times \text{unit_cost}$$

so any record whose provider-reported total disagrees with the recomputed figure beyond a tolerance is a data-quality signal caught at the boundary, not a rounding artifact discovered in finance’s spreadsheet.

Step 1 — Write the converter helpers

Provider exports serialize cost as a JSON float or a string, and strict=True rejects both for a Decimal field on purpose. Two BeforeValidator hooks run ahead of the strict type check: one converts any numeric or string cost into an exact Decimal(str(v)) (never Decimal(float), which would re-import the rounding error you are trying to avoid), and one normalizes every timestamp to timezone-aware UTC so mixed offsets never skew a billing_period rollup. Wrapping each in an Annotated alias keeps the model declaration below clean.

from datetime import datetime, timezone
from decimal import Decimal, InvalidOperation
from typing import Annotated, Any

from pydantic import BeforeValidator


def to_decimal(v: Any) -> Decimal:
    """Coerce numeric or string cost inputs into an exact Decimal."""
    if isinstance(v, Decimal):
        return v
    if isinstance(v, (int, float, str)):
        try:
            return Decimal(str(v))  # str() first to avoid binary-float drift
        except InvalidOperation as exc:
            raise ValueError(f"non-numeric cost value: {v!r}") from exc
    raise TypeError("cost must be numeric or string")


def to_utc(v: Any) -> datetime:
    """Normalize timestamp inputs to timezone-aware UTC datetimes."""
    if isinstance(v, datetime):
        return v if v.tzinfo else v.replace(tzinfo=timezone.utc)
    if isinstance(v, str):
        try:
            dt = datetime.fromisoformat(v.replace("Z", "+00:00"))
        except ValueError as exc:
            raise ValueError(f"invalid ISO-8601 timestamp: {v!r}") from exc
        return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc)
    raise TypeError("timestamp must be str or datetime")


DecimalCost = Annotated[Decimal, BeforeValidator(to_decimal)]
UtcDatetime = Annotated[datetime, BeforeValidator(to_utc)]

Step 2 — Assemble the strict record contract

Now declare the model. ConfigDict(strict=True, extra="forbid") stops Pydantic from silently coercing a stringified integer into an int and blocks arbitrary metadata injection — a common billing-pipeline corruption vector. cost_basis is required and constrained to an enum so a payload that never states whether its figure is unblended or amortized fails loudly rather than being averaged into the ledger. The money and timestamp fields use the annotated aliases from Step 1, and a model_validator(mode="after") proves the record is internally consistent: total_cost must equal usage_quantity × unit_cost, or the record is rejected. This mirrors the strict-typing discipline applied to cost allocation tags.

from typing import Dict, Literal

from pydantic import BaseModel, ConfigDict, Field, model_validator

CostBasis = Literal["unblended", "amortized"]
TOLERANCE = Decimal("0.0001")  # absorb float-to-Decimal drift in usage_quantity


class CostLineItem(BaseModel):
    model_config = ConfigDict(strict=True, extra="forbid")

    resource_id: str = Field(min_length=1, max_length=128)
    cluster_name: str = Field(pattern=r"^[a-z0-9-]+$")
    metric_name: str = Field(min_length=1)
    cost_basis: CostBasis
    currency: str = Field(pattern=r"^[A-Z]{3}$", default="USD")
    usage_quantity: float = Field(ge=0.0)
    unit_cost: DecimalCost = Field(ge=0)
    total_cost: DecimalCost = Field(ge=0)
    tags: Dict[str, str] = Field(default_factory=dict)
    timestamp: UtcDatetime

    @model_validator(mode="after")
    def reconcile_total(self) -> "CostLineItem":
        expected = Decimal(str(self.usage_quantity)) * self.unit_cost
        if abs(expected - self.total_cost) > TOLERANCE:
            raise ValueError(
                f"cost reconciliation failed for {self.resource_id}: "
                f"expected {expected}, got {self.total_cost}"
            )
        return self

Field validation proves each value is well-formed; the reconciliation validator catches the discrepancies that come from pagination truncation, provider-side rounding policy, or manual ledger edits — failing fast before a corrupted figure reaches quota enforcement.

Step 3 — Ingest with dead-letter routing

Wrap parsing and validation so a JSONDecodeError and a ValidationError are handled distinctly — the first is a transport problem, the second a contract problem — and both send the raw payload to a dead-letter queue with a machine-readable reason. ValidationError.json() emits field paths, error types, and offending values, which is exactly the structured context an on-call engineer needs instead of a bare stack trace.

import json
import logging
from dataclasses import dataclass

from pydantic import ValidationError

logger = logging.getLogger("finops.billing_validator")


@dataclass
class DeadLetter:
    raw: str
    reason: str


def ingest_billing_payload(raw_json: str) -> CostLineItem | DeadLetter:
    """Parse and validate one payload; quarantine anything that fails."""
    try:
        payload = json.loads(raw_json)
    except json.JSONDecodeError as exc:
        logger.error("malformed JSON payload: %s", exc)
        return DeadLetter(raw=raw_json, reason=f"JSONDecodeError: {exc}")
    try:
        return CostLineItem.model_validate(payload)
    except ValidationError as exc:
        logger.error("schema validation failed: %s", exc.json())
        return DeadLetter(raw=raw_json, reason=exc.json())


if __name__ == "__main__":
    sample = json.dumps({
        "resource_id": "db-prod-analytics-01",
        "cluster_name": "prod-analytics",
        "metric_name": "vcpu_hours",
        "cost_basis": "unblended",
        "usage_quantity": 12.5,
        "unit_cost": "0.096",
        "total_cost": "1.20",
        "tags": {"team": "data-platform"},
        "timestamp": "2026-06-02T00:00:00Z",
    })
    result = ingest_billing_payload(sample)
    print(type(result).__name__, "->", result)

The flow below traces a raw payload through parsing and validation to either the cost ledger or the dead-letter queue.

Ingest path: raw payload through json.loads and the validity gate to the cost ledger, or to the dead-letter queueA raw JSON payload is parsed by json.loads; a JSONDecodeError diverts it to the dead-letter queue while a parsed dict passes to CostLineItem.model_validate. A "Schema and cost math valid?" gate sends ValidationError failures to the dead-letter queue and valid records on to a typed CostLineItem, then to the cost ledger and quota enforcement. The dead-letter queue routes to remediation and replay.parsed dictvalidJSONDecodeErrorValidationErrorRaw JSONpayloadjson.loadsCostLineItem.model_validate()Schema & costmath valid?TypedCostLineItemCost ledger &quota enforcementDead-letterqueueRemediation& replay

Verification

Confirm the validator both accepts a clean record and quarantines a broken one before wiring it into any attribution job.

  1. Assert the happy path returns a typed record. A well-formed payload must reconcile and return a CostLineItem with Decimal money and UTC time.

    good = ingest_billing_payload(sample)
assert isinstance(good, CostLineItem)
assert isinstance(good.total_cost, Decimal)
assert good.timestamp.tzinfo is not None
print("validated:", good.resource_id, good.total_cost, good.currency)

    Expected output:

    validated: db-prod-analytics-01 1.20 USD

  2. Assert a math mismatch is quarantined. Break the reconciliation and confirm the record is dead-lettered, not accepted.

    bad = json.loads(sample)
bad["total_cost"] = "9.99"          # 12.5 * 0.096 = 1.20, not 9.99
result = ingest_billing_payload(json.dumps(bad))
assert isinstance(result, DeadLetter)
assert "reconciliation failed" in result.reason
print("quarantined:", result.reason[:60])

  3. Confirm strict typing bites. Feed a stringified usage_quantity ("12.5" instead of 12.5); under strict=True it must fail rather than silently coerce, proving the boundary is real.

Gotchas & Edge Cases

  • Credits and refunds are negative cost. The ge=0 bounds above reject Savings Plan true-ups and refunds that legitimately arrive as negative total_cost. If your export carries them, drop the ge=0 constraint on the cost fields, keep it on usage_quantity, and add a record_kind enum (charge / credit) so the reconciliation branch handles sign explicitly.
  • strict=True rejects provider floats until the BeforeValidator runs. The order matters: a BeforeValidator executes before strict type enforcement, so it is what lets a "0.096" string through as a Decimal. Retyping a field to plain Decimal without the annotated alias will start rejecting real payloads.
  • Decimal(0.1) is not Decimal("0.1"). Always stringify first. Constructing a Decimal directly from a binary float re-imports the exact rounding error the Decimal migration exists to eliminate.
  • extra="forbid" breaks on provider schema drift. When a provider adds a new field, a forbidding model raises on every record until you update it. That is the intended fail-loud behavior, but pair it with dead-letter routing so drift degrades to a quarantine queue rather than a pipeline outage — the same posture as normalizing provider billing exports into a unified schema.
  • Naive timestamps skew period rollups. A timestamp without an offset is assumed UTC by to_utc, which is safe only if the provider truly emits UTC. Providers that emit local time need an explicit source timezone before canonicalization, or a day’s spend lands in the wrong billing_period.
  • Blended figures fail single-tenant attribution. Requiring cost_basis does not make a blended figure correct for per-tenant chargeback; it only makes the dimension explicit. Reconcile against unblended or amortized, never blended.

Frequently Asked Questions

Why coerce money to Decimal instead of just using float?

Because IEEE-754 floats cannot represent most decimal fractions exactly, and the tiny per-value error compounds across millions of sub-cent line items into real reconciliation drift. Decimal gives exact base-10 arithmetic, so a summed ledger matches the invoice to the cent. The one rule is to construct it from a string — Decimal(str(v)) — because Decimal(0.1) inherits the float’s error while Decimal("0.1") does not.

strict=True rejects my provider’s float costs — how do I still accept them?

Attach a BeforeValidator to the field via Annotated[Decimal, BeforeValidator(to_decimal)]. Before-validators run ahead of Pydantic’s strict type check, so your function receives the raw float or string, returns an exact Decimal, and the strict checker then sees a value that is already the correct type. This keeps strict mode’s guarantees for every other field while accepting the money representations providers actually send.

How do I collect all validation errors instead of failing on the first?

Pydantic already does this — a single ValidationError aggregates every field failure in one payload, and exc.errors() returns them as a list of dicts with loc, type, and msg. There is no need to validate field by field. For a whole batch, catch per record and append each DeadLetter so one bad row never aborts the run, exactly as the ingest function above does.

Should validation live in the pipeline or the database schema?

Both, at different depths. The Pydantic contract is the ingestion gate that rejects structurally or arithmetically broken records before they are ever written, giving you machine-readable errors and dead-letter routing. Database CHECK constraints and NOT NULL are the last-line backstop for whatever bypasses the pipeline. Relying on the database alone means the first signal of corruption is a failed INSERT with no context about the source payload.

How do I feed validated records straight into quota enforcement?

A validated CostLineItem is already typed and reconciled, so it can be aggregated per cluster_name and compared against a budget with no further defensive checks. Sum the total_cost values into a rolling window and hand the result to your quota policy layer to translate normalized cost signals into hard and soft limits. Because validation guarantees the math, a breach alert reflects real spend rather than a parsing artifact.

Back to: Schema Validation for Billing Data