Parsing JSON Responses in Python: A Builder's Guide to Reliable API Integration

A JSON parser that works on your laptop and fails at 3 a.m. is the most expensive kind of code you can ship, because it fails quietly. Part of the Getting Started with Python APIs for Builders guide, this page covers how to turn an untrusted response body into a typed object you can bill against, log about, and hand to the rest of your service without hedging.

The distinction that matters commercially is between a decode failure and a contract failure. A decode failure is loud: the bytes were not JSON, json.loads raises, and you find out immediately. A contract failure is silent: the bytes decoded fine, the shape changed, and data.get("results", []) now returns an empty list forever because the vendor renamed the key to items. Your sync job reports success, your dashboard shows zero new records, and nobody notices for eleven days. Every technique on this page exists to convert the second class of failure into the first.

Calling .json() and indexing into the result works for exactly as long as the upstream API is stable and your traffic is low. Once you are parsing a few million payloads a month on a container with a memory limit, three things hurt at once: decoding cost, object-graph memory, and the retries you waste on responses that were never going to succeed. This guide fixes all three.

Prerequisites

You need Python 3.11 or newer, because the routing code below leans on structural pattern matching — match over a dataclass with capture patterns and guards — to keep the envelope logic flat instead of nesting five conditionals. Install a deliberately small set of packages:

python -m venv .venv && . .venv/bin/activate
pip install \
  "httpx>=0.28" \
  "pydantic>=2.9,<3" \
  "orjson>=3.10" \
  "tenacity>=9.0" \
  "structlog>=24.4" \
  "pytest>=8.3" \
  "pytest-asyncio>=0.24" \
  "respx>=0.22"

Use httpx rather than requests for anything new. It exposes the same ergonomic .json() method, but it gives you a real async client, per-request timeouts that actually cover connect and read separately, and HTTP/2 when the upstream supports it. The full comparison lives in httpx vs requests for async; if you are still on the synchronous library, the patterns here port over one for one and the base transport concerns are covered in making HTTP requests with the requests library.

Every knob comes from the environment. Nothing below hardcodes a host, a key, or a size limit, because the same image has to run against a vendor sandbox in staging and the real endpoint in production.

export UPSTREAM_BASE_URL="https://api.vendor.example"
export UPSTREAM_API_KEY="sk_live_replace_me"
export UPSTREAM_TIMEOUT_SECONDS="10"
export UPSTREAM_CONNECT_TIMEOUT_SECONDS="3"
export UPSTREAM_MAX_BODY_BYTES="8388608"
export UPSTREAM_ENVELOPE_KEY="results"
export UPSTREAM_MAX_RETRIES="4"
export PARSE_STRICT="true"

Step 1: Know the Envelope Before You Write the Parser

Two payload conventions dominate, and they fail in opposite ways. REST endpoints wrap collections in an envelope — {"results": [...], "next": "..."} or {"data": [...], "meta": {...}} — and signal failure with the HTTP status line. GraphQL returns a body shaped exactly like the query you sent, nests everything under data, and answers 200 OK even when half your resolvers blew up. The failure information lives in a sibling errors array that a naive client never reads.

That asymmetry decides your parser. Against REST you can branch on status first and treat a 2xx body as structurally trustworthy enough to hand to a validator. Against GraphQL the status tells you almost nothing, so you must inspect the body before you decide whether the call succeeded. Getting this backwards is the single most common reason a GraphQL integration reports 100 percent uptime while delivering partial data. The wider architectural trade-offs are laid out in understanding REST vs GraphQL; what matters here is that they need genuinely different unwrapping code.

Two smaller conventions catch people out. Some APIs return a bare top-level array with no envelope, which breaks any code calling .get() on the decoded value — a list has no .get, so a line that looks defensive raises AttributeError. Others return 204 No Content for an empty result set, and a 204 body is zero bytes, so .json() raises a decode error on a completely successful response. Handle both explicitly.

Step 2: Decode Defensively with httpx

Calling .json() on a response you have not inspected is the most common cause of a confusing traceback in an integration codebase, because upstream error pages are usually HTML. When a load balancer in front of your vendor returns a 502 with a branded error page, .json() raises a decode error whose message mentions a character at position zero, and you waste twenty minutes looking for a JSON bug that does not exist.

Check three things before decoding: the status, the content type, and the body size. The size check matters more than people expect — a paginated endpoint that normally returns 40 KB will occasionally return 200 MB when someone passes per_page=100000, and decoding that inside a 512 MB container takes the whole process down.

# integration/fetch.py
from __future__ import annotations

import json
import os
from dataclasses import dataclass

import httpx


class PayloadTooLarge(RuntimeError):
    """Upstream returned more bytes than we are willing to decode."""


class NotJson(RuntimeError):
    """Upstream returned a 2xx response that was not a JSON document."""


@dataclass(frozen=True)
class Decoded:
    status: int
    body: object | None
    raw: bytes


def _limits() -> httpx.Timeout:
    return httpx.Timeout(
        float(os.getenv("UPSTREAM_TIMEOUT_SECONDS", "10")),
        connect=float(os.getenv("UPSTREAM_CONNECT_TIMEOUT_SECONDS", "3")),
    )


def build_client() -> httpx.AsyncClient:
    return httpx.AsyncClient(
        base_url=os.getenv("UPSTREAM_BASE_URL", ""),
        timeout=_limits(),
        headers={
            "Authorization": f"Bearer {os.getenv('UPSTREAM_API_KEY', '')}",
            "Accept": "application/json",
        },
        http2=os.getenv("UPSTREAM_HTTP2", "true").lower() == "true",
        limits=httpx.Limits(max_connections=int(os.getenv("UPSTREAM_MAX_CONNS", "20"))),
    )


async def decode(client: httpx.AsyncClient, path: str, **params: str) -> Decoded:
    response = await client.get(path, params=params)
    raw = response.content

    max_bytes = int(os.getenv("UPSTREAM_MAX_BODY_BYTES", "8388608"))
    if len(raw) > max_bytes:
        raise PayloadTooLarge(f"{len(raw)} bytes from {path}, limit {max_bytes}")

    if response.status_code == 204 or not raw:
        return Decoded(status=response.status_code, body=None, raw=raw)

    media_type = response.headers.get("content-type", "").split(";")[0].strip()
    if not media_type.endswith("json"):
        raise NotJson(f"{media_type or 'no content-type'} from {path}")

    try:
        return Decoded(status=response.status_code, body=json.loads(raw), raw=raw)
    except json.JSONDecodeError as exc:
        preview = raw[:200].decode("utf-8", errors="replace")
        raise NotJson(f"undecodable body from {path}: {preview}") from exc

Three details earn their keep. Testing media_type.endswith("json") accepts the real-world variants — application/vnd.api+json, application/problem+json, text/json — without accepting text/html. Decoding response.content rather than response.text skips a decode-then-encode round trip, since json.loads takes bytes directly. And the 200-character preview in the exception means the log line tells you what actually arrived.

Notice what this function does not do: it does not raise on a 4xx. A 402 or a 429 has a JSON body you want to read, and raise_for_status() throws that body away. Gate on status after you have the decoded body in hand.

Step 3: Unwrap the Envelope with a match Statement

Now route the decoded state. A match statement over the status and body shape reads better than a nest of if branches, and — more importantly — it forces you to write a case _ arm, which is where the shapes you did not anticipate end up instead of silently falling through.

# integration/unwrap.py
from __future__ import annotations

import os
from typing import Any

from integration.fetch import Decoded


class UpstreamError(RuntimeError):
    def __init__(self, status: int, detail: str, retryable: bool) -> None:
        super().__init__(f"{status}: {detail}")
        self.status = status
        self.detail = detail
        self.retryable = retryable


def _detail(body: Any) -> str:
    match body:
        case {"error": {"message": str(message)}}:
            return message
        case {"error": str(message)} | {"detail": str(message)} | {"message": str(message)}:
            return message
        case {"errors": [{"message": str(message)}, *_]}:
            return message
        case _:
            return "no error detail in body"


def records(decoded: Decoded) -> list[dict[str, Any]]:
    """Return the record list, or raise a classified UpstreamError."""
    envelope_key = os.getenv("UPSTREAM_ENVELOPE_KEY", "results")

    match decoded:
        case Decoded(status=204) | Decoded(body=None):
            return []
        case Decoded(status=int(code)) if code >= 500:
            raise UpstreamError(code, _detail(decoded.body), retryable=True)
        case Decoded(status=429):
            raise UpstreamError(429, _detail(decoded.body), retryable=True)
        case Decoded(status=int(code)) if code >= 400:
            raise UpstreamError(code, _detail(decoded.body), retryable=False)
        case Decoded(body={"errors": [_, *_]}):
            raise UpstreamError(200, _detail(decoded.body), retryable=False)
        case Decoded(body=list() as rows):
            return rows
        case Decoded(body={**envelope}) if envelope_key in envelope:
            found = envelope[envelope_key]
            if not isinstance(found, list):
                raise UpstreamError(200, f"{envelope_key} was {type(found).__name__}", retryable=False)
            return found
        case _:
            keys = sorted(decoded.body)[:6] if isinstance(decoded.body, dict) else type(decoded.body).__name__
            raise UpstreamError(200, f"unrecognised shape, keys={keys}", retryable=False)

The last arm is the important one. It refuses to guess. If the vendor renames results to items, this raises with unrecognised shape, keys=['items', 'next'] instead of returning an empty list, and your alerting catches a contract change on the first request rather than on the first customer complaint. That single behavioural choice — fail loudly on an unknown shape — is what separates an integration you can trust from one you merely hope about.

The errors arm sits deliberately after the status arms so a GraphQL response carrying both data and a partial errors array counts as a failure. If your business logic genuinely tolerates partial results, split that case: read data, count the errors, and emit a metric through whatever you set up in monitoring and logging Python APIs so somebody can see the degradation.

Step 4: Validate Into Typed Models with Pydantic v2

A list of dictionaries is not a data contract. It is a promise you have no way to check. Pydantic v2 turns that promise into an assertion that runs in Rust at a cost you can measure, and it gives you a ValidationError with a machine-readable location for every field that drifted.

Use model_validate_json and hand it the raw bytes when you can. It parses and validates in one pass inside pydantic-core, skipping the intermediate Python dictionary entirely — roughly 40 percent faster than json.loads followed by model_validate, and materially lighter on memory because those throwaway dicts are never allocated. When you have already decoded (because you needed to route on the envelope first, as above), model_validate on the extracted list is the right call.

# integration/models.py
from __future__ import annotations

import os
from datetime import datetime
from decimal import Decimal
from typing import Annotated, Literal

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

STRICT = os.getenv("PARSE_STRICT", "true").lower() == "true"


class Order(BaseModel):
    model_config = ConfigDict(
        extra="ignore",
        strict=STRICT,
        populate_by_name=True,
        frozen=True,
    )

    id: str
    external_ref: str | None = Field(default=None, alias="externalRef")
    status: Literal["pending", "paid", "refunded", "cancelled"]
    amount_cents: Annotated[int, Field(ge=0, alias="amountCents")]
    currency: Annotated[str, Field(min_length=3, max_length=3)]
    created_at: datetime = Field(alias="createdAt")

    @property
    def amount(self) -> Decimal:
        return Decimal(self.amount_cents) / 100


ORDERS = TypeAdapter(list[Order])


def parse_orders(rows: list[dict]) -> tuple[list[Order], list[dict]]:
    """Validate every row; return the good ones and a report of the bad ones."""
    good: list[Order] = []
    rejected: list[dict] = []
    for index, row in enumerate(rows):
        try:
            good.append(Order.model_validate(row))
        except ValidationError as exc:
            rejected.append(
                {
                    "index": index,
                    "id": row.get("id"),
                    "problems": [
                        {"field": ".".join(str(p) for p in e["loc"]), "kind": e["type"]}
                        for e in exc.errors(include_url=False)
                    ],
                }
            )
    return good, rejected

Row-by-row validation instead of ORDERS.validate_python(rows) is a deliberate trade. The TypeAdapter is faster and rejects the whole batch on a single bad record, which is what you want for a payment reconciliation job where partial data is worse than no data. The loop above is what you want for an ingest pipeline where one malformed row out of 5,000 should not stall the sync. Pick per use case and be explicit about which one you chose; the ambiguity is where bugs live.

Three configuration choices deserve defending. extra="ignore" drops fields you did not model, which keeps memory flat when a vendor adds twelve new attributes you do not care about. strict=STRICT is switchable by environment so you can run strict in production and lax while you are still discovering the real shape in staging — in strict mode a "42" string will not silently become the integer 42, which is exactly the coercion that lets a type change slip through unnoticed. And frozen=True makes the model hashable and prevents a downstream function from mutating parsed data in place, which is worth it for the debugging time alone. Discriminated unions, custom validators, and nested model performance are covered in depth in validating JSON with Pydantic v2, and the same models become your response schemas the moment you expose this data through FastAPI — which is also what makes them show up correctly when documenting APIs with OpenAPI.

Log the rejected list as structured data, never as a formatted string. Shipping loc, type, and msg as separate keys lets you query for "how many int_parsing failures on amountCents this week" instead of grepping — and that query is what tells you a vendor changed a type before it becomes a billing dispute.

Step 5: Route Failures Instead of Retrying Everything

Retrying blindly is expensive in two currencies. It burns your own compute, and on a metered upstream it burns quota you paid for on requests that were never going to succeed. A 401 will still be a 401 after four exponential backoffs; all you have bought is a 45-second delay before the error surfaces.

Classify first, then retry only the retryable class. The UpstreamError above already carries a retryable flag, so wiring tenacity to respect it is four lines.

# integration/client.py
from __future__ import annotations

import os

import httpx
from tenacity import (
    AsyncRetrying,
    retry_if_exception,
    stop_after_attempt,
    wait_exponential_jitter,
)

from integration.fetch import build_client, decode
from integration.models import Order, parse_orders
from integration.unwrap import UpstreamError, records


def _should_retry(exc: BaseException) -> bool:
    match exc:
        case UpstreamError() as err:
            return err.retryable
        case httpx.TimeoutException() | httpx.ConnectError() | httpx.RemoteProtocolError():
            return True
        case _:
            return False


async def fetch_orders(path: str, **params: str) -> tuple[list[Order], list[dict]]:
    retrying = AsyncRetrying(
        retry=retry_if_exception(_should_retry),
        stop=stop_after_attempt(int(os.getenv("UPSTREAM_MAX_RETRIES", "4"))),
        wait=wait_exponential_jitter(
            initial=float(os.getenv("UPSTREAM_BACKOFF_INITIAL", "0.5")),
            max=float(os.getenv("UPSTREAM_BACKOFF_MAX", "20")),
        ),
        reraise=True,
    )
    async with build_client() as client:
        async for attempt in retrying:
            with attempt:
                rows = records(await decode(client, path, **params))
        return parse_orders(rows)

wait_exponential_jitter matters more than the retry count. Without jitter, every worker that hit the same 503 wakes up at the same millisecond and hits the recovering upstream simultaneously, which is how a brief blip becomes a sustained outage. The deeper patterns — budget caps, circuit breaking, honouring Retry-After — are in retrying failed HTTP requests with tenacity, and if the class you keep hitting is a 429 the diagnosis path is debugging 429 too many requests errors rather than more retries.

Configuration Reference

Every setting is read at call time, so a container restart is enough to change behaviour. The defaults are safe for local development; the recommendations assume a paid upstream and a container with a memory limit.

Two of these interact in a way that bites. UPSTREAM_MAX_RETRIES multiplied by UPSTREAM_BACKOFF_MAX must stay below the timeout of whatever called you. Four attempts with a 20-second cap can hold a request open for the better part of a minute, and if that call sits inside an HTTP handler your own client times out first, leaving the retry loop running against a socket nobody is reading. Retry inside background jobs, not inside request handlers.

UPSTREAM_MAX_BODY_BYTES deserves an actual calculation rather than a guess. Decoded JSON typically occupies fifteen to twenty-five times its wire size as Python objects, so an 8 MB cap can allocate roughly 160 MB at peak. On a 512 MB container serving three concurrent syncs, that is an out-of-memory kill. Either lower the cap or switch that endpoint to the incremental approach in handling large JSON payloads with streaming.

Gotchas and Failure Modes

Consuming the body before you verify a webhook signature. The symptom is a signature check that fails for every request in production and passes in every test. Stripe and GitHub sign the exact bytes they sent; await request.json() in FastAPI gives you a re-serialized structure whose key order and whitespace differ, so an HMAC over it never matches. Read await request.body() first, verify against those bytes, then decode. The full receiver pattern is in processing webhooks with Python and the signature specifics in verifying Stripe webhook signatures.

Floats where money should be. The symptom is a reconciliation report that is off by a cent on a few hundred rows. json.loads maps every JSON number to a Python float, so 19.99 becomes 19.989999999999998 before Pydantic ever sees it. Model monetary fields as integer minor units, as Order.amount_cents does above. If the vendor genuinely sends decimal strings, type the field as Decimal and let Pydantic parse the string — never let a float touch a currency amount.

Silent truncation by .get() chains. The symptom is a pipeline that reports success while writing nothing. body.get("data", {}).get("items", []) returns an empty list for a renamed key, a null value, and a completely restructured payload alike, and every one of those is indistinguishable from a genuinely empty page. Reserve .get() with a default for fields that are legitimately optional, and raise on a missing structural key.

Duplicate keys collapsing without warning. JSON permits repeated object keys and json.loads silently keeps the last one. Vendors that build payloads by string concatenation emit duplicates surprisingly often, and the record you receive is not the record they think they sent. If you suspect this, pass object_pairs_hook to json.loads and raise when a key repeats — it costs a few microseconds and once caught a vendor bug that would otherwise have taken a week to reproduce.

Trusting Content-Length for the size check. The symptom is a memory limit that never triggers. Compressed responses report the compressed length in the header, so a 3 MB gzip body that expands to 90 MB passes any check written against Content-Length. Measure len(response.content) after decompression, as the code above does.

Timezone-naive datetimes. The symptom is off-by-hours bugs in usage reports around month boundaries. If the vendor sends 2026-07-23T14:00:00 with no offset, Pydantic produces a naive datetime that compares unequal to every aware one in your database. Normalise at the boundary: reject naive timestamps in strict mode, or attach UTC explicitly in a field validator.

Verification

Prove the parser handles bad input, not just good input. respx intercepts httpx at the transport layer, which lets you assert on the exact failure modes above without a network call or a vendor sandbox account.

# tests/test_parsing.py
import json

import httpx
import pytest
import respx

from integration.fetch import NotJson, decode, build_client
from integration.unwrap import UpstreamError, records

BASE = "https://api.vendor.example"


@pytest.fixture(autouse=True)
def _env(monkeypatch):
    monkeypatch.setenv("UPSTREAM_BASE_URL", BASE)
    monkeypatch.setenv("UPSTREAM_ENVELOPE_KEY", "results")


@pytest.mark.asyncio
@respx.mock
async def test_html_error_page_is_rejected():
    respx.get(f"{BASE}/orders").mock(
        return_value=httpx.Response(502, text="<html>bad gateway</html>",
                                    headers={"content-type": "text/html"})
    )
    async with build_client() as client:
        with pytest.raises(NotJson):
            await decode(client, "/orders")


@pytest.mark.asyncio
@respx.mock
async def test_renamed_envelope_key_raises_instead_of_returning_empty():
    respx.get(f"{BASE}/orders").mock(
        return_value=httpx.Response(200, json={"items": [{"id": "o_1"}], "next": None})
    )
    async with build_client() as client:
        with pytest.raises(UpstreamError, match="unrecognised shape"):
            records(await decode(client, "/orders"))


@pytest.mark.asyncio
@respx.mock
async def test_204_is_an_empty_page_not_an_error():
    respx.get(f"{BASE}/orders").mock(return_value=httpx.Response(204))
    async with build_client() as client:
        assert records(await decode(client, "/orders")) == []


@pytest.mark.asyncio
@respx.mock
async def test_graphql_partial_errors_are_treated_as_failure():
    body = {"data": {"orders": []}, "errors": [{"message": "field resolver timeout"}]}
    respx.get(f"{BASE}/graphql").mock(
        return_value=httpx.Response(200, content=json.dumps(body),
                                    headers={"content-type": "application/json"})
    )
    async with build_client() as client:
        with pytest.raises(UpstreamError, match="resolver timeout"):
            records(await decode(client, "/graphql"))

The second test is the one that pays for itself. It is the regression guard for the exact silent-empty-list failure that costs eleven days of missing data, and it runs in under a millisecond. Wire the suite into CI alongside the rest of your coverage; the broader approach is in testing Python APIs with pytest, and mocking technique specifically in mocking external APIs with respx.

For a fast manual check against the real endpoint, pipe the response through a shape summary rather than eyeballing it:

curl -s -H "Authorization: Bearer $UPSTREAM_API_KEY" \
  "$UPSTREAM_BASE_URL/orders?per_page=1" \
  | python -c 'import json,sys; d=json.load(sys.stdin); print(type(d).__name__, sorted(d)[:8] if isinstance(d,dict) else len(d))'

Cost and Performance at Scale

Parsing CPU is almost never your bill. Memory headroom is. Here is a measured comparison on a 2 MB response containing 5,000 order records, on a single shared vCPU:

Read that chart with the memory column in mind, because it inverts the ranking. The four in-memory strategies all peak between 41 MB and 96 MB of resident Python objects for a 2 MB wire payload; the streaming approach peaks at about 9 MB. If you run four workers on a 512 MB container and each one occasionally handles a 2 MB page, the json.loads plus model_validate path can spike past 380 MB before garbage collection catches up, and the container gets killed. Moving that one endpoint to incremental parsing lets you stay on the 512 MB instance at roughly seven dollars a month instead of jumping to 2 GB at around twenty-five. That is a two-hundred-dollar-a-year decision made entirely by parse strategy, and it is invisible until the first out-of-memory restart.

Now the CPU side, honestly. At three million parses a month of a typical 60 KB payload, switching from decode-then-validate to model_validate_json saves roughly 0.75 milliseconds per call, or about 38 minutes of CPU across the whole month. On a shared vCPU billed near three cents an hour that is under two dollars. Real but not decisive. The genuinely expensive line item is wasted work: retrying a 401 four times triples your outbound request count against a metered vendor for zero successful responses, and at a tenth of a cent per upstream call across three million requests that is three thousand dollars of quota bought and thrown away. Classification, not micro-optimisation, is where the margin is. Fold both numbers into the model described in calculating cost per API request.

Two levers move the number materially. A five-minute cache on a catalogue endpoint that ninety percent of your traffic hits removes ninety percent of your parsing outright — see caching Python API responses with Redis. And field selection at the source, ?fields=id,status,amount where the vendor supports it, is cheaper still, because bytes you never receive cost nothing to decode, allocate, or validate. A sparse-fieldset parameter routinely cuts payloads by seventy percent.

One niche deserves a warning: LLM APIs. Model output that claims to be JSON arrives wrapped in Markdown fences, truncated at a token limit, or subtly malformed, and every attempt costs money — so a blind retry loop is the fastest way to burn a budget. Validate into a strict model and re-prompt with the validation error rather than resending the same request. The economics are worked through in controlling LLM API costs in production, within the wider patterns of automating AI workflows with Python APIs.

FAQ

How much does careless JSON parsing actually cost at a million requests a month? The decode itself is trivial — a million 60 KB payloads is roughly nine minutes of CPU, well under a dollar. The damage comes from the two failure modes around it. Retrying non-retryable errors four times turns a million upstream calls into three million, which at a tenth of a cent per call is two thousand dollars of pure waste. And a memory spike that forces you from a 512 MB instance to a 2 GB one adds about eighteen dollars a month per container.

Is Pydantic validation worth the CPU on a high-throughput ingest path? Yes, and it is cheaper than people assume if you use it correctly. model_validate_json on raw bytes runs about 40 percent faster than decoding to a dict first, so the marginal cost over plain json.loads is roughly one millisecond per 60 KB payload. Weigh that against a single silent type change flowing into your database: the cleanup is days of engineering plus whatever the wrong data cost your customers. Skip validation only on data you never store and never bill against.

A vendor is about to change their response shape. How do I migrate without downtime? Model both shapes and accept either during the transition. Add the new field as optional with an alias, keep the old one optional too, and use a model validator that requires exactly one of them to be present. Deploy that tolerant version first, watch your rejection metric fall to zero, then remove the old branch in a later release. This is the client-side mirror of the discipline in versioning and evolving public APIs, and the deciding factor is whether you log rejected rows well enough to know when the migration is complete.

Should I use json.loads or response.json()? Use response.json() when you already trust the response, because it handles encoding detection for you. Use json.loads(response.content) when you want control — which is most production code — because it lets you check content type and body size first, pass object_pairs_hook to catch duplicate keys, and include a body preview in the error message. Swap in orjson.loads for the same interface at roughly two and a half times the speed once payloads get large.

Why does my rejected-row count spike after a vendor deploy, and what should I do about it? Because their schema changed and your strict models caught it, which is the system working. Alert on the rejection rate rather than the absolute count, since a batch-size change alone will move the count. When the rate jumps, pull the field and kind values from your structured logs to see exactly which attribute drifted, and prefer widening the model over disabling strict mode — turning strict off globally to unblock a deploy converts every future schema change back into a silent corruption you will not notice.

Related

• Validating JSON with Pydantic v2 — discriminated unions, custom validators, and the model configuration this guide only sketches.
• Handling Large JSON Payloads with Streaming — the incremental approach that keeps peak memory near 9 MB when a payload will not fit.
• Debugging 401 Unauthorized API Errors — diagnose the credential failures your classifier should never retry.
• Understanding REST vs GraphQL — why the two envelope styles need different unwrapping code in the first place.
• Structured Logging with structlog — ship rejected-row reports as queryable fields instead of formatted strings.