Validating Sensor Payloads with Pydantic

A payload validator’s job is narrower than it first appears: it decides whether a message is structurally a reading, not whether the reading is good. Get that boundary wrong in either direction and the pipeline suffers — too permissive and a string "NaN" propagates into a database column, too strict and the validator silently deletes the extreme values that environmental monitoring exists to capture. This page builds a Pydantic v2 model that draws the line correctly, coerces units at the boundary, and routes every rejection to a dead-letter record that says why. It sits immediately before the registry join described in sensor metadata and device registry management.

What Belongs in the Schema and What Does Not

Three questions separate a schema concern from a quality concern.

Can the pipeline represent this value at all? A latitude of 412, a timestamp of "soon", a measurement of null — these cannot be stored in a typed column, so they are schema failures.

Is the value physically possible for this instrument? A relative humidity of 140% cannot exist; a PM2.5 of 900 µg/m³ can, during a severe wildfire. The first is a schema failure because no valid instrument state produces it. The second is a quality question.

Is the value plausible given context? A temperature 8 °C above every neighbouring sensor is suspicious but perfectly representable. This is never a schema concern — it belongs to anomaly detection, which has the context to judge it and the flag vocabulary to record its judgement.

The practical rule: the validator’s bounds should be the instrument’s physical limits from its datasheet, widened slightly, not the range you expect to see. A sensor that can only report −40 to +85 °C should reject 200 °C, because that value means a decode error rather than a hot day.

Where the schema boundary sits, and what lies either side of it Flow diagram separating structural rejection from quality flagging: decode, then schema validation which rejects unrepresentable values, then QC flagging which keeps extreme but possible values with a flag. Where the schema boundary sits, and what lies either side of it Decode bytes → typed fields its own error class Schema guard representable? reject → dead letter Registry join identity, location quarantine unknowns QC flagging plausible? flag, never delete Bounds in the schema come from the instrument datasheet; bounds on plausibility come from the QC stage. Confusing them deletes real episodes.
The extreme reading during a wildfire is structurally fine and statistically unusual — which is exactly why the two stages are separate.

Production-Ready Implementation 

# python 3.11 · pydantic==2.7.1
from __future__ import annotations

from datetime import datetime, timedelta, timezone
from typing import Annotated, Literal

from pydantic import BaseModel, ConfigDict, Field, ValidationError, field_validator

MAX_FUTURE = timedelta(minutes=5)      # allow small clock skew, reject time travel
MAX_AGE = timedelta(days=30)           # older than this is a replay, not telemetry


class SensorReading(BaseModel):
    """One decoded environmental reading, at the ingest boundary.

    Bounds are the *instrument's* physical limits, not the expected range: the
    job here is to reject values that no working sensor can produce, and to let
    every extreme-but-possible value through to QC flagging with its flag intact.
    """

    model_config = ConfigDict(extra="forbid", str_strip_whitespace=True)

    device_id: Annotated[str, Field(min_length=3, max_length=64, pattern=r"^[A-Za-z0-9:_-]+$")]
    observed_at: datetime
    metric: Literal["pm25", "temperature_c", "humidity_pct", "pressure_hpa"]
    value: float
    battery_v: float | None = Field(default=None, ge=0, le=6)

    @field_validator("observed_at")
    @classmethod
    def _sane_time(cls, v: datetime) -> datetime:
        if v.tzinfo is None:
            raise ValueError("observed_at must be timezone-aware (UTC)")
        now = datetime.now(timezone.utc)
        if v > now + MAX_FUTURE:
            raise ValueError(f"observed_at is {v - now} in the future")
        if v < now - MAX_AGE:
            raise ValueError(f"observed_at is {now - v} old — replay or clock fault")
        return v.astimezone(timezone.utc)

    @field_validator("value")
    @classmethod
    def _physically_possible(cls, v: float, info) -> float:
        limits = {
            "pm25": (0.0, 2000.0),          # sensor saturates well below this
            "temperature_c": (-60.0, 90.0),  # beyond the datasheet range either way
            "humidity_pct": (0.0, 100.0),
            "pressure_hpa": (300.0, 1100.0),
        }
        metric = info.data.get("metric")
        if metric is None:                   # metric already failed; do not mask that error
            return v
        lo, hi = limits[metric]
        if not (lo <= v <= hi):
            raise ValueError(f"{metric}={v} outside instrument range [{lo}, {hi}]")
        return v

Two configuration choices carry weight. extra="forbid" turns an unexpected field into an error rather than silently discarding it — which is how you find out that a firmware update started sending pm10 alongside pm25 instead of discovering it a quarter later. And rejecting naive datetimes at the boundary means the timezone discipline described in handling timezone drift is enforced by the type system rather than by convention.

The dead-letter path is the other half of the implementation. A rejection that is not recorded is data loss with extra steps:

from dataclasses import dataclass


@dataclass(frozen=True)
class Rejected:
    raw: dict
    reason: str
    field: str | None


def parse(raw: dict) -> tuple[SensorReading | None, Rejected | None]:
    """Validate one payload. Never raises: a failure is a value, not an exception."""
    try:
        return SensorReading.model_validate(raw), None
    except ValidationError as exc:
        first = exc.errors()[0]
        field = ".".join(str(p) for p in first["loc"]) or None
        return None, Rejected(raw=raw, reason=first["msg"], field=field)

Returning the failure instead of raising it matters in a streaming consumer: an exception that escapes the message handler either kills the consumer or, worse, is caught by a broad except two frames up that logs nothing and acknowledges the message anyway.

Reject or flag: the same value, judged twice Matrix of six candidate readings showing whether the schema guard rejects them and whether the QC stage flags them, making the division of responsibility concrete. Reject or flag: the same value, judged twice Schema QC stage Ends up as PM2.5 = 612 µg/m³ (wildfire) accept flag anomalous published, flagged PM2.5 = 41 000 µg/m³ reject — dead letter Humidity = 100 % accept flag saturated published, flagged Humidity = 140 % reject — dead letter Timestamp naive reject — dead letter Unknown field pm10 reject — dead letter (firmware change)
Every rejection is a value no working instrument could produce. Everything else survives with a flag.

Parameter Tuning Guide

Metric Instrument limits Reject outside Typical real range Flag, do not reject
PM2.5 (µg/m³) 0–1 000 (optical) 0–2 000 2–60 above 150 during smoke
Temperature (°C) −40 to +85 −60 to +90 −10 to +38 above 45 in direct sun
Relative humidity (%) 0–100 0–100 25–95 exactly 0 or 100 (saturated)
Pressure (hPa) 300–1 100 300–1 100 950–1 040 below 950 in a storm
Dissolved oxygen (mg/L) 0–20 0–25 4–12 below 2 (hypoxia — real)
Battery (V) 0–6 0–6 3.2–4.2 below 3.3 (imminent failure)

The last column is the discipline this table exists to enforce. Every value in it is real, alarming and worth keeping — and every one of them would be deleted by a validator whose bounds were set from “what we usually see”.

Validation throughput, single core Bar chart of models validated per second for per-message validation, batch validation through a TypeAdapter, and the arrival rate of a large sensor network for comparison. Validation throughput, single core 0 100000 200000 104000 Per message 210000 Batch (TypeAdapter) 1400 Network arrival rate records / second
Validation is not the bottleneck at any realistic environmental network size — the third bar is a 20 000-sensor fleet at one-minute cadence.

Verification and Testing

Test the boundary, not the happy path. The cases that matter are the ones where reject-versus-flag is a judgement call.

# python 3.11 · pydantic==2.7.1 · pytest==8.2.0
import pytest
from datetime import datetime, timezone


def _payload(**over):
    base = {
        "device_id": "eui-70b3d5",
        "observed_at": datetime.now(timezone.utc).isoformat(),
        "metric": "pm25",
        "value": 18.4,
    }
    base.update(over)
    return base


def test_extreme_but_real_value_is_accepted():
    reading, rejected = parse(_payload(value=612.0))   # severe wildfire smoke
    assert rejected is None and reading.value == 612.0


def test_impossible_value_is_rejected_with_a_reason():
    reading, rejected = parse(_payload(value=41000.0))
    assert reading is None
    assert "outside instrument range" in rejected.reason
    assert rejected.field == "value"


def test_naive_timestamp_is_rejected():
    _, rejected = parse(_payload(observed_at="2026-08-01T10:00:00"))
    assert rejected is not None and "timezone-aware" in rejected.reason


def test_unknown_field_is_rejected_not_ignored():
    _, rejected = parse(_payload(pm10=22.0))
    assert rejected is not None and rejected.field == "pm10"

In production, the metric to watch is the rejection rate per reason code. A steady 0.1% unknown_field rate is provisioning noise; the same rate jumping to 40% overnight is a firmware rollout, and it will be obvious within minutes if the reason code is a labelled counter rather than a log line.

Gotchas

float accepts "NaN" and "Infinity" from JSON. Python’s json module parses both by default, and Pydantic’s float coercion accepts them. A NaN that reaches a database column poisons every average computed from it. Add an explicit math.isfinite check, or configure the JSON parser with parse_constant to raise.

Validating before decoding hides the real error. If a LoRaWAN payload fails to decode, the resulting dict is missing fields and the validator reports “field required” — which sends whoever is debugging to the wrong layer. Decode first, and let the decoder produce its own error class.

A permissive model with every field | None validates nothing. It is tempting when payload shapes vary by model, and it converts every schema failure into a silent null. Use a discriminated union on the sensor model instead, so each variant states exactly what it requires.

Bounds copied from the expected range rather than the datasheet. Worth repeating because it is the mistake with the highest cost: a validator tuned to typical values deletes exactly the episodes — the smoke event, the flood, the hypoxic night — that justify the network’s existence.

FAQ

Should validation reject a reading or flag it?

Reject only what is structurally unusable — a missing timestamp, an unparseable number, a coordinate outside the globe. Anything that is well-formed but implausible is a quality question, not a schema question, and belongs in the QC flagging stage where it gets a flag code and stays in the dataset. A validator that drops a genuine 400 µg/m³ reading during a wildfire has destroyed the most important data of the year.

Is Pydantic fast enough for a high-rate stream?

Yes. Pydantic 2 validates in compiled Rust and handles roughly 100 000 simple models per second per core, which is far beyond the arrival rate of any environmental network. If you are genuinely constrained, validate at batch granularity with TypeAdapter over a list rather than per message — that alone roughly doubles throughput.

How do I validate a payload whose fields vary by sensor model?

Use a discriminated union keyed on the model field. Each variant gets its own model with its own ranges and required fields, and Pydantic picks the right one from the discriminator without a chain of try/except. It also produces a far better error message than a single permissive model with every field optional.