Gowtham Potureddi

Posted on Jul 14

Pydantic for Data Engineering: Schema Validation in ETL & Pipeline Contracts

#python #sql #interview #dataengineering

pydantic is the Python library every data engineer eventually reaches for when a Kafka payload with a missing field silently corrupts a downstream aggregate, when a CSV row with a malformed date reaches the warehouse and skews the reporting table for a whole quarter, or when an internal API service ships a schema change that its downstream consumer parses as garbage instead of failing loudly at the wire. Every DE eventually validates a runtime payload; knowing when to enforce a schema at the ingress boundary versus letting the pipeline crash-fast on the downstream operation, when to reach for a full BaseModel versus a lightweight @dataclass, and when to bypass validation via model_construct in a hot loop is what separates a senior operator from a mid-level one. Pydantic v2 rewrote the internals in Rust (pydantic-core), delivering a 5-50× speedup over v1 across parse-and-validate operations, and made typed models the default surface for FastAPI, LangChain, and dozens of other Python libraries that DE teams touch daily.

The tour walks the five pillars every DE needs to keep straight in 2026 — (1) BaseModel + Field constraints (gt=0, max_length=100, pattern=r"^[A-Z]{3}$") plus custom validators (@field_validator for per-field transformation, @model_validator(mode="after") for cross-field business rules) that enforce domain invariants at parse time, (2) pipeline contracts at Kafka / API / CSV / file boundaries using model_validate_json for the parse-and-validate one-shot plus DLQ routing for rejected messages plus discriminated-union versioning via Literal["v1"] tags, (3) the v2 Rust-core performance story where a typical model parse is 5-20× faster than v1, the JSON parse-plus-validate hot path is 10-50× faster, and validation cost drops out of the top 5 CPU items in most pipelines, (4) contract-first design via Model.model_json_schema() for JSON Schema export to schema registries (Confluent, Apicurio) so downstream consumers in Java / Go / TypeScript generate typed clients from the same wire contract, and (5) the decision matrix vs dataclasses (no runtime validation), attrs (like dataclass with more features), TypedDict (type-checker-only hint), and SQLModel (Pydantic + SQLAlchemy fusion for CRUD-heavy apps). Every section ships a teaching block followed by a Solution-Tail interview answer — code, a step-by-step trace, an output table, then a concept-by-concept breakdown of why it works — so you leave with the two-line skeleton and the reason it wins.

When you want hands-on reps immediately after reading, drill the SQL practice library → for schema-validation patterns that show up at the SQL boundary, sharpen SQL aggregation drills → for the downstream aggregate correctness that depends on clean ingress, and layer the broader SQL optimization surface → covering 450+ DE-focused problems on data quality, contracts, and pipeline correctness.

On this page

Why Pydantic matters for DE in 2026
BaseModel + Field validators
Pipeline contracts + Kafka payloads
Pydantic v2 performance + Rust core
Contracts as first-class + dialect matrix
Cheat sheet — Pydantic recipe list
Frequently asked questions
Practice on PipeCode

1. Why Pydantic matters for DE in 2026

The `pydantic` mental model — validate once at the ingress boundary, trust the typed model through the rest of the pipeline

The one-sentence invariant: Pydantic's job in a data pipeline is to enforce a data contract at the ingress boundary — the Kafka consumer, the HTTP endpoint, the CSV row, the config file — so that invalid data is rejected loudly (to a DLQ, a 422 response, an error log) instead of leaking downstream to corrupt aggregates, feature stores, and BI dashboards; validate once at the boundary, trust the typed Model instance through every subsequent transform, and skip validation only in tight hot loops where you can prove the values are already trusted. Every reporting incident that traces back to "how did that malformed row get through" started with a pipeline that lacked a boundary validator.

Where Pydantic actually shows up in DE pipelines.

Kafka payload parsing. The consumer receives raw bytes; OrderEvent.model_validate_json(msg.value) parses + validates in one call. Malformed messages route to a dead-letter topic; valid ones flow downstream as typed OrderEvent instances.
FastAPI request bodies. Every DE-adjacent internal API (dbt job trigger, feature-store fetch, pipeline-status query) uses FastAPI + Pydantic; the framework auto-generates 422 responses on schema violations and OpenAPI docs from the models.
CSV / JSON parsing at ETL ingest. Row-level validation before insert to warehouse. A row with total: "not a number" is rejected with a specific field-level error instead of ballooning into a Snowflake COPY failure with a cryptic message.
Airflow XCom. Task A returns a dict; Task B parses it via a Pydantic model, so a contract mismatch surfaces immediately rather than as a NoneType error four tasks downstream.
Feature-store row schemas. Model the feature row; validate before write to online store (Redis / DynamoDB) so downstream ML serving never reads a corrupted feature.
Configuration files. BaseSettings from pydantic-settings for typed env vars — the DE app fails to start if DATABASE_URL is missing or malformed, not three lines into a query.
dbt post-processors. Python scripts that read dbt models via SQLAlchemy, transform, and write derived tables validate the intermediate shape with Pydantic.
CDC apply-loops. Debezium emits change events; the apply job parses them via a versioned Pydantic model that survives schema evolution.
LangChain / LLM pipelines. Every structured-output feature (tool call, JSON mode, function calling) uses Pydantic models to define the wire schema.

The four "why did this bug happen" root causes Pydantic prevents.

The silent-drop bug. A field missing from a JSON payload is silently None in a raw dict; downstream code does if x['field'] > 0 and hits KeyError. Pydantic raises at parse time with the exact field name.
The type-coercion bug. A CSV column reads "42" (string) but downstream expects int; ad-hoc int(row['x']) scattered through the codebase misses one row. Pydantic coerces once at parse, or refuses via strict=True.
The contract-drift bug. Producer service adds a new required field; consumer keeps working (or silently misparses). Pydantic model on both sides fails-fast on both ends of the wire.
The malformed-JSON bug. A truncated or malformed Kafka payload reaches downstream processing; json.loads succeeds partially or raises unhandled. Pydantic's model_validate_json catches both parse and schema errors under one exception.

What senior interviewers actually probe when Pydantic shows up.

Do you know the BaseModel vs dataclass decision? Pydantic for untrusted input; dataclass for internal trusted state.
Do you know Field constraints? gt, ge, lt, le, min_length, max_length, pattern, default, default_factory, alias, frozen, exclude.
Do you know @field_validator vs @model_validator? Per-field transformation vs cross-field invariants.
Do you know v2's coercion behavior? By default, "42" → 42 (coerce); strict=True refuses.
Do you know model_validate vs model_validate_json? dict → model vs bytes/str → model in one call.
Do you know the DLQ pattern? Malformed messages route to a dead-letter topic with the error captured; retain payload for debugging.
Do you know schema versioning via Literal? Discriminated union tag lets one endpoint accept multiple schema versions atomically.
Do you know when to skip validation? model_construct for internal trusted handoffs where the values are already validated upstream.
Do you know v2's Rust core? pydantic-core is a Rust library; validation is 5-50× faster than v1; the excuse "validation is slow" no longer applies.
Do you know how to export JSON Schema? Model.model_json_schema(); publish to a schema registry for polyglot consumers.

The reading discipline — how a senior looks at a pipeline for Pydantic gaps.

Step 1 — find every ingress boundary. Kafka consumer, HTTP endpoint, CSV read, file parse, DB read of untrusted source. Every one needs a Pydantic model at the boundary.
Step 2 — check what happens on malformed input. Silent None? Exception three functions later? Or clean rejection at the boundary?
Step 3 — check the downstream shape. After the boundary, is everything typed with the Pydantic model or does it degrade to dict[str, Any]?
Step 4 — check for coercion surprises. "42" → 42 may not be what you want; strict=True documents intent.
Step 5 — check contract versioning. If the producer added a field, does the consumer break loudly or silently ignore?

Worked example — the silent NoneType bug that Pydantic catches

Detailed explanation. A pipeline consumes a Kafka topic of user profile update events. The producer sometimes emits messages missing the email field for anonymous users. Downstream code accesses event['email'].lower() expecting a string, hits AttributeError: 'NoneType' object has no attribute 'lower', and the whole worker crashes. The right fix is a Pydantic model at the boundary that declares email as Optional[str] = None and rejects only the truly malformed messages.

Question. Show the naive raw-dict handler failing, then the Pydantic handler that either rejects malformed input or produces a typed model with email: Optional[str].

Input. Three Kafka messages:

#	Payload (JSON)	Expected outcome
1	`{"id": 42, "email": "a@x.com"}`	Valid — process
2	`{"id": 43, "email": null}`	Valid but no email — process
3	`{"id": "not-int", "email": "b@x.com"}`	Invalid — DLQ

Code.

# NAIVE (buggy): raw dict handler
import json
for message in kafka_consumer:
    payload = json.loads(message.value)
    email_lower = payload["email"].lower()   # AttributeError on message #2
    process(payload["id"], email_lower)

# FIXED: Pydantic model at boundary
from pydantic import BaseModel, ValidationError
from typing import Optional

class UserEvent(BaseModel):
    id: int
    email: Optional[str] = None

for message in kafka_consumer:
    try:
        event = UserEvent.model_validate_json(message.value)
    except ValidationError as e:
        dead_letter_queue.send(message.value, error=e.errors())
        continue
    email_lower = event.email.lower() if event.email else None
    process(event.id, email_lower)

Step-by-step explanation.

Naive handler reads message #2 with email: null. payload["email"] returns None. None.lower() raises AttributeError — the worker crashes and the whole consumer group stalls.
Pydantic model declares email: Optional[str] = None — nullable field, default None.
Message #1 parses cleanly, event.email = "a@x.com".
Message #2 parses cleanly, event.email = None (explicit null accepted).
Message #3 fails to parse — id: "not-int" can't coerce to int under default settings that expect valid int strings only for the primitive; Pydantic raises ValidationError with a structured error list. The message goes to the DLQ with the exact field-level cause.
Downstream code guards event.email with if event.email else None — the None case is now visible and explicit rather than implicit.

Output.

Message	Naive result	Pydantic result
#1 (valid)	processed	processed
#2 (email null)	AttributeError crash	processed with email=None
#3 (id malformed)	crash on downstream int()	DLQ with clear error
Worker state after all 3	crashed	still consuming

Rule of thumb. Every Kafka / HTTP / CSV ingress point needs a Pydantic model. The naive dict[str, Any] approach is a bug factory.

Worked example — the type-coercion bug on CSV int fields

Detailed explanation. A CSV loader reads orders.csv where the total column is a string in the source ("$25.00"), converts to float in Python via float(row['total']). One row has total = "" (empty). float("") raises ValueError. Pydantic's declarative field + field_validator for the currency-symbol strip fixes both problems.

Question. Show the buggy loader, then the Pydantic loader that handles empty strings and currency symbols in one place.

Input. CSV rows:

id,total,currency
1,$25.00,USD
2,,USD
3,$40.50,USD

Code.

# BUGGY
import csv
for row in csv.DictReader(open("orders.csv")):
    total = float(row["total"].lstrip("$"))   # ValueError on row 2

# FIXED
from pydantic import BaseModel, Field, field_validator

class Order(BaseModel):
    id: int
    total: float = Field(ge=0)
    currency: str = Field(pattern=r"^[A-Z]{3}$")

    @field_validator("total", mode="before")
    @classmethod
    def strip_currency_symbol(cls, v):
        if isinstance(v, str):
            v = v.lstrip("$").strip()
            if v == "":
                return 0.0   # or None if field is Optional
        return v

for row in csv.DictReader(open("orders.csv")):
    order = Order.model_validate(row)
    process(order)

Step-by-step explanation.

Naive code assumes every total is a currency-prefixed decimal string. Empty string fails.
Pydantic Order.total is declared as float with ge=0 constraint.
@field_validator("total", mode="before") runs BEFORE type coercion — receives raw string, strips $ and whitespace, converts empty to 0.0.
Pydantic then coerces the resulting string to float (or accepts already-float).
ge=0 runs after coercion; rejects negative totals as domain violations.
currency is validated against the ISO-4217 3-letter pattern.

Output.

Row	Naive	Pydantic
1 (`$25.00, USD`)	25.0	Order(id=1, total=25.0, currency='USD')
2 (`, USD`)	ValueError	Order(id=2, total=0.0, currency='USD')
3 (`$40.50, USD`)	40.5	Order(id=3, total=40.5, currency='USD')

Rule of thumb. Coercion logic lives inside @field_validator(mode="before"); the field type declares intent; the constraint (ge=0) enforces domain rule. Never scatter cleanup code across the pipeline.

Worked example — the contract-drift bug and schema versioning

Detailed explanation. A producer service ships v1 events with {id, name, email}. A month later, v2 adds tenant_id as a required field. The consumer still deploys the old model; new v2 events fail parsing. Discriminated union with Literal tag lets one consumer handle both versions atomically.

Question. Design a consumer that accepts v1 and v2 payloads simultaneously via a discriminated union on the version tag.

Input.

{"version": "v1", "id": 42, "name": "alice", "email": "a@x.com"}
{"version": "v2", "id": 43, "name": "bob", "email": "b@x.com", "tenant_id": 100}

Code.

from pydantic import BaseModel, Field
from typing import Literal, Union, Annotated

class UserEventV1(BaseModel):
    version: Literal["v1"]
    id: int
    name: str
    email: str

class UserEventV2(BaseModel):
    version: Literal["v2"]
    id: int
    name: str
    email: str
    tenant_id: int

UserEvent = Annotated[
    Union[UserEventV1, UserEventV2],
    Field(discriminator="version"),
]

# Parse
import json
class Envelope(BaseModel):
    event: UserEvent

for msg in kafka_consumer:
    envelope = Envelope.model_validate({"event": json.loads(msg.value)})
    if isinstance(envelope.event, UserEventV2):
        process_with_tenant(envelope.event)
    else:
        process_legacy(envelope.event)

Step-by-step explanation.

Both UserEventV1 and UserEventV2 have a version field declared as Literal["v1"] or Literal["v2"] — Pydantic uses this literal as the discriminator.
Annotated[Union[...], Field(discriminator="version")] tells Pydantic: "peek at the version field; pick the matching model."
Parser reads a message, checks version tag, dispatches to the right model.
Consumer code can use isinstance to branch on version.
Adding v3 later: define UserEventV3, add to union — no changes to existing code paths.

Output.

Payload version	Parsed as	Fields available
v1	UserEventV1	id, name, email
v2	UserEventV2	id, name, email, tenant_id
v3 (future)	UserEventV3	id, name, email, tenant_id, region

Rule of thumb. Every event / payload should carry a version tag; use Literal + discriminated union to accept multiple versions atomically without branching in application code.

Common beginner mistakes

Using raw dict[str, Any] everywhere instead of typing the ingress boundary.
Coercion logic scattered — int(...), float(...), str(...) cleanup across the codebase instead of one @field_validator.
Forgetting Optional[T] = None for nullable fields — Pydantic requires explicit optionality.
Assuming email: Optional[str] alone allows None — must also provide default = None.
Using v1 syntax in v2 code (@validator deprecated; use @field_validator).
Not catching ValidationError — application crashes on malformed input instead of DLQ.
Forgetting to declare mode="before" when the raw input needs preprocessing.
Setting strict=True and being surprised when "42" doesn't coerce to 42.
Trying to serialise a model with .dict() (v1) — v2 uses .model_dump().

`pydantic` interview question on choosing between raw dict, dataclass, and BaseModel

A senior interviewer often opens with: "You're building the ingest layer for a new Kafka topic. Walk me through the decision — raw dict, @dataclass, TypedDict, or Pydantic BaseModel? What's the criteria, and what's the fallback if validation cost dominates?"

Solution Using ingress-boundary Pydantic + internal dataclass hand-off + model_construct escape hatch

from dataclasses import dataclass
from pydantic import BaseModel, ValidationError, ConfigDict
from typing import Optional

# INGRESS BOUNDARY — validated at parse
class KafkaOrderEvent(BaseModel):
    model_config = ConfigDict(frozen=True)   # immutable
    id: int
    customer_id: int
    total: float
    currency: str
    ts: str

# INTERNAL PIPELINE — trusted, cheap
@dataclass(frozen=True, slots=True)
class ProcessedOrder:
    id: int
    customer_id: int
    total_usd: float

# HOT LOOP ESCAPE — skip validation entirely
def rebuild_from_snapshot(row) -> KafkaOrderEvent:
    # data is already trusted (came from our own validated store)
    return KafkaOrderEvent.model_construct(
        id=row.id, customer_id=row.customer_id,
        total=row.total, currency=row.currency, ts=row.ts,
    )

# Ingress
for msg in kafka_consumer:
    try:
        event = KafkaOrderEvent.model_validate_json(msg.value)
    except ValidationError as e:
        dlq.send(msg.value, error=e.errors())
        continue
    # Convert to internal dataclass — cheap, typed, immutable
    processed = ProcessedOrder(
        id=event.id, customer_id=event.customer_id,
        total_usd=convert_to_usd(event.total, event.currency),
    )
    downstream(processed)

Step-by-step trace.

Layer	Type	Validation cost	When
Kafka wire	JSON bytes	—	at ingress
KafkaOrderEvent	Pydantic BaseModel	5-50 μs per message	at ingress
ProcessedOrder	dataclass	~0 (allocation only)	internal handoff
Snapshot rebuild	model_construct	~0 (no validation)	trusted rebuild only

The three-tier design — Pydantic at boundary, dataclass internal, model_construct escape hatch — gives you the validation guarantee where it matters and the raw-Python speed everywhere else.

Output:

Approach	Cost per msg	Correctness	Best for
Raw dict	1 μs	None (silent bugs)	never
@dataclass	2 μs	Type hints only (no runtime)	internal state
TypedDict	1 μs	Type-check-only	dict shape hints
Pydantic BaseModel	5-50 μs	Full runtime validation	ingress boundary
model_construct	2 μs	Type hints, no validation	trusted internal

Why this works — concept by concept:

Pydantic at ingress — the boundary is where untrusted data crosses into the pipeline; validate once and produce a typed model. Downstream code trusts the model.
dataclass for internal state — after validation, the pipeline manipulates its own domain objects; runtime validation is redundant and costly. @dataclass(frozen=True, slots=True) is cheap and typed.
model_construct escape hatch — for the rare case where you rebuild a Pydantic model from already-trusted data (a warehouse snapshot, a cache read), bypassing validation is safe and 10-20× faster than model_validate.
ConfigDict(frozen=True) — makes the model immutable, preventing accidental mutation downstream; catches bugs at write time.
DLQ on ValidationError — malformed messages retain their payload and error details for debugging; the consumer never crashes on bad input.
Cost — ingress validation is O(1) per message with a 5-50 μs constant; internal transforms are O(1) allocation. Total per-message cost is dominated by downstream I/O (DB write, HTTP call), not by Pydantic.

SQL
Topic — SQL
SQL practice library

Practice →

SQL Topic — aggregation SQL aggregation drills

Practice →

2. BaseModel + Field validators

`pydantic BaseModel` + `Field` constraints + `@field_validator` + `@model_validator` — the four primitives that cover 90% of DE validation

The mental model in one line: BaseModel gives you a typed class with automatic parse-and-validate; Field(gt=0, min_length=1, pattern=...) adds declarative constraints per field; @field_validator("col", mode="before"|"after") adds custom per-field transformation or validation; @model_validator(mode="after") adds cross-field business rules that inspect the fully-parsed model — combine the four and you can express any DE schema contract without writing a single if outside the model.

Slot 1 — BaseModel basics.

Inherit from BaseModel. Every field becomes a runtime-validated attribute.
Type annotations declare intent. id: int, name: str, age: int | None = None.
Instantiate normally. User(id=42, name="Alice") — validation runs at __init__.
Parse from dict. User.model_validate({"id": 42, "name": "Alice"}).
Parse from JSON. User.model_validate_json(b'{"id": 42, "name": "Alice"}').
Serialize to dict. user.model_dump().
Serialize to JSON. user.model_dump_json().

Slot 2 — Field constraints (numeric).

gt=0 — greater than.
ge=0 — greater than or equal.
lt=100 — less than.
le=100 — less than or equal.
multiple_of=5 — must be divisible by 5.
allow_inf_nan=False — reject inf / NaN.

Slot 3 — Field constraints (string).

min_length=1 — non-empty.
max_length=100 — cap.
pattern=r"^[A-Z]{3}$" — regex.
strict=True — disable coercion for this field.
to_lower=True — auto-lowercase (via str_to_lower model config).

Slot 4 — Field metadata.

default=0 — default value.
default_factory=lambda: uuid.uuid4() — computed default per instance.
alias="userId" — accept userId from wire, expose as user_id internally.
description="..." — feeds into JSON Schema.
examples=[...] — sample values for OpenAPI docs.
frozen=True — this field can't be reassigned after construction.
exclude=True — don't include in model_dump().

Slot 5 — @field_validator — the per-field escape hatch.

Signature. @field_validator("field_name") decorator on a classmethod.
Modes. mode="before" runs BEFORE Pydantic's type coercion; mode="after" runs after (default).
Return. Return the (possibly transformed) value or raise ValueError.
Common uses. Currency-symbol strip, timezone normalization, string trimming, enum lookup.

@field_validator("email", mode="after")
@classmethod
def lowercase_email(cls, v: str) -> str:
    return v.lower()

@field_validator("total", mode="before")
@classmethod
def strip_currency_symbol(cls, v):
    if isinstance(v, str):
        v = v.lstrip("$").strip()
    return v

Slot 6 — @model_validator — the cross-field escape hatch.

Signature. @model_validator(mode="before"|"after") decorator on a classmethod (before) or instance method (after).
mode="before" — receives raw input dict; can transform before parsing.
mode="after" — receives the fully-parsed model; can inspect all fields and raise.
Common uses. "total > 0 requires quantity > 0"; "end_date >= start_date"; conditional required fields.

class Order(BaseModel):
    total: float = Field(ge=0)
    quantity: int
    discount_pct: float = 0.0

    @model_validator(mode="after")
    def check_business_rules(self):
        if self.total > 0 and self.quantity <= 0:
            raise ValueError("total > 0 requires quantity > 0")
        if self.discount_pct > 0 and self.total == 0:
            raise ValueError("discount only valid on non-zero total")
        return self

Slot 7 — nested models.

Automatic recursion. address: Address inside a Customer model — Pydantic recursively parses and validates the nested dict.
Optional nested. address: Optional[Address] = None.
List of nested. items: list[LineItem].
Dict of nested. by_id: dict[int, LineItem].

Slot 8 — coercion vs strict.

Default (coerce). "42" → 42; 1 → True; "2026-07-14" → date(2026,7,14).
ConfigDict(strict=True) — model-wide; refuses coercion.
Field(..., strict=True) — field-scoped; refuses coercion just for that field.
Use strict for. Wire protocols with schema registries where you want fail-fast on wrong types.
Use coerce for. CSV / form-encoded / URL-param sources where everything arrives as string.

Slot 9 — Optional[T] and defaults.

name: Optional[str] — type is nullable but still required in the payload.
name: Optional[str] = None — nullable AND optional (default None).
name: str | None = None — same as above, PEP 604 syntax.
name: str = Field(default=None) — invalid; type mismatch.
name: list[int] = [] — DANGEROUS in v1, safe in v2 (v2 copies per instance).

Slot 10 — default_factory for computed defaults.

UUID. id: UUID = Field(default_factory=uuid.uuid4).
Timestamp. created_at: datetime = Field(default_factory=lambda: datetime.now(UTC)).
Empty list per instance. tags: list[str] = Field(default_factory=list) — safer than = [].

Slot 11 — model config.

ConfigDict(str_strip_whitespace=True) — auto-trim strings.
ConfigDict(str_to_lower=True) — auto-lowercase.
ConfigDict(extra="forbid") — reject unknown fields.
ConfigDict(extra="ignore") — silently drop unknown (default).
ConfigDict(extra="allow") — keep unknown as attributes.
ConfigDict(populate_by_name=True) — accept both alias and field name.
ConfigDict(from_attributes=True) — parse from any object with matching attrs (formerly orm_mode).

Slot 12 — the six most common Field patterns for DE.

Positive amount. total: Decimal = Field(gt=0).
Percentage. discount: float = Field(ge=0, le=100).
Non-empty string. name: str = Field(min_length=1).
ISO currency. currency: str = Field(pattern=r"^[A-Z]{3}$").
Email (import EmailStr). email: EmailStr (requires pip install pydantic[email]).
UUID. id: UUID = Field(default_factory=uuid.uuid4).

Common beginner mistakes

Using v1 @validator — deprecated; use @field_validator.
Forgetting @classmethod decorator on @field_validator.
Mixing up mode="before" and mode="after" — before runs on raw input, after runs on coerced value.
Trying to modify self in mode="after" model validator — must return self after mutation.
Using mutable default (= [], = {}) instead of default_factory=list — v1 issue, v2 safer but still preferred.
Assuming Optional[T] means "defaultable" — it means "nullable"; you still need = None.
Using extra="allow" and getting surprised by attribute-not-declared errors.
Not using EmailStr / HttpUrl from pydantic — reinventing regex validators.

Worked example — a complete Order model with all four primitives

Detailed explanation. A production-grade Order model that validates every field with declarative constraints, transforms currency symbols, enforces cross-field invariants, and produces clean output.

Question. Write an Order model that accepts messy JSON from a shopping-cart service and produces a fully-validated typed instance.

Input.

{
  "id": 42,
  "customer_id": 1001,
  "total": "$125.50",
  "currency": "usd",
  "quantity": 3,
  "discount_pct": 10.0,
  "coupon_code": "  SAVE10  ",
  "email": "Alice@Example.COM",
  "created_at": "2026-07-12T14:32:00Z"
}

Code.

from pydantic import BaseModel, Field, EmailStr, ConfigDict, field_validator, model_validator
from datetime import datetime

class Order(BaseModel):
    model_config = ConfigDict(
        str_strip_whitespace=True,
        frozen=True,
    )

    id: int = Field(ge=1)
    customer_id: int = Field(ge=1)
    total: float = Field(ge=0)
    currency: str = Field(pattern=r"^[A-Z]{3}$")
    quantity: int = Field(ge=1)
    discount_pct: float = Field(ge=0, le=100)
    coupon_code: str | None = Field(default=None, max_length=20)
    email: EmailStr
    created_at: datetime

    @field_validator("total", mode="before")
    @classmethod
    def strip_currency(cls, v):
        if isinstance(v, str):
            v = v.lstrip("$").strip()
            if v == "":
                return 0
        return v

    @field_validator("currency", mode="before")
    @classmethod
    def upper_currency(cls, v: str) -> str:
        return v.upper() if isinstance(v, str) else v

    @field_validator("email", mode="after")
    @classmethod
    def lower_email(cls, v: str) -> str:
        return v.lower()

    @model_validator(mode="after")
    def check_business_rules(self):
        if self.discount_pct > 0 and self.total == 0:
            raise ValueError("discount only valid on non-zero total")
        if self.total > 10000 and self.coupon_code is None:
            raise ValueError("high-value orders require coupon justification")
        return self

order = Order.model_validate({
    "id": 42, "customer_id": 1001, "total": "$125.50", "currency": "usd",
    "quantity": 3, "discount_pct": 10.0, "coupon_code": "  SAVE10  ",
    "email": "Alice@Example.COM", "created_at": "2026-07-12T14:32:00Z",
})

Step-by-step explanation.

model_config = ConfigDict(str_strip_whitespace=True) — every string field is auto-trimmed at parse.
frozen=True — model instance is immutable after construction.
Fields declare types and constraints. id: int = Field(ge=1) — must be positive integer.
@field_validator("total", mode="before") — receives raw "$125.50", strips $, returns "125.50" string.
Pydantic then coerces "125.50" → 125.50 (float type).
@field_validator("currency", mode="before") — upper-cases "usd" → "USD".
Pydantic then validates against the ^[A-Z]{3}$ pattern — passes.
@field_validator("email", mode="after") — after EmailStr validation, lowercase the domain.
str_strip_whitespace=True handles " SAVE10 " → "SAVE10".
@model_validator(mode="after") runs after all fields validated; checks cross-field business rules.
Instance is fully typed, immutable, ready for downstream use.

Output.

Order(
    id=42,
    customer_id=1001,
    total=125.50,
    currency='USD',
    quantity=3,
    discount_pct=10.0,
    coupon_code='SAVE10',
    email='alice@example.com',
    created_at=datetime(2026, 7, 12, 14, 32, 0, tzinfo=timezone.utc),
)

Rule of thumb. A production Order / Event model layers config → field types → Field constraints → @field_validator for per-field transforms → @model_validator for cross-field rules. Every layer is optional; use as needed.

Worked example — refusing coercion with strict mode

Detailed explanation. A wire protocol with a schema registry expects fields to arrive in the exact declared type. Silent coercion ("42" → 42) hides producer bugs; strict mode refuses.

Question. Build a WireEvent model that refuses type coercion for external inputs.

Code.

from pydantic import BaseModel, ConfigDict

class WireEvent(BaseModel):
    model_config = ConfigDict(strict=True)
    id: int
    total: float

# Coerce would-have-worked
try:
    WireEvent.model_validate({"id": "42", "total": "100.5"})
except Exception as e:
    print(e)
# ValidationError: id -> Input should be a valid integer [type=int_type]

Rule of thumb. Use strict=True on wire protocols where producer contract is strict; use default (coerce) on CSV / URL params where everything is a string.

Worked example — `model_validate_json` vs `model_validate`

Detailed explanation. Bytes/str payloads should use model_validate_json to avoid a manual json.loads step; dict payloads use model_validate.

Code.

# From dict (already parsed)
user = User.model_validate({"id": 42, "name": "Alice"})

# From JSON bytes (parse and validate in one call — faster)
user = User.model_validate_json(b'{"id": 42, "name": "Alice"}')

# From JSON string
user = User.model_validate_json('{"id": 42, "name": "Alice"}')

Rule of thumb. model_validate_json is measurably faster than json.loads(...) + model_validate(...) because Pydantic-core (Rust) does both in one pass.

`pydantic BaseModel` interview question on nested models and lists

A senior interviewer asks: "Design a Pydantic model for an e-commerce order with N line items. Each item has an sku, quantity, unit price. Validate the total = sum of line items. Show the full model."

Solution Using nested BaseModel + list[Item] + model_validator for total invariant

from pydantic import BaseModel, Field, model_validator
from typing import Annotated

class LineItem(BaseModel):
    sku: str = Field(pattern=r"^[A-Z0-9-]+$")
    quantity: int = Field(ge=1)
    unit_price: float = Field(ge=0)

    @property
    def line_total(self) -> float:
        return self.quantity * self.unit_price

class Order(BaseModel):
    id: int
    customer_id: int
    items: list[LineItem] = Field(min_length=1)
    total: float = Field(ge=0)

    @model_validator(mode="after")
    def check_total_matches_items(self):
        expected = sum(item.line_total for item in self.items)
        # Allow small floating-point tolerance
        if abs(expected - self.total) > 0.01:
            raise ValueError(f"total={self.total} does not match items sum={expected}")
        return self

order = Order.model_validate({
    "id": 42, "customer_id": 1001,
    "items": [
        {"sku": "WIDGET-A", "quantity": 2, "unit_price": 25.0},
        {"sku": "GADGET-B", "quantity": 1, "unit_price": 75.5},
    ],
    "total": 125.50,
})

Step-by-step trace.

Step	Action	Result
1	Parse `id`, `customer_id`	Coerced to int
2	Parse `items` — a list	Each dict recursively validated as LineItem
3	LineItem.sku pattern check	Passes for WIDGET-A and GADGET-B
4	LineItem.quantity ge=1, unit_price ge=0	Passes
5	Parse `total`	125.50
6	`@model_validator(mode="after")` runs	Sum items = 225 + 175.5 = 125.5, matches
7	Order instance ready	Fully typed with nested LineItem[]

Output:

Field	Value
id	42
customer_id	1001
items	[LineItem(sku='WIDGET-A', quantity=2, unit_price=25.0), LineItem(sku='GADGET-B', quantity=1, unit_price=75.5)]
total	125.50

Why this works — concept by concept:

Nested BaseModel for LineItem — declares the shape of each item; Pydantic recursively validates.
list[LineItem] field — Pydantic parses each dict in the list into a LineItem.
Field(min_length=1) on the list — enforces at least one item.
@property for line_total — computed attribute; not validated but usable in cross-field logic.
@model_validator(mode="after") for cross-field invariant — sums all items post-parse; compares to declared total; raises on mismatch beyond floating tolerance.
Floating-point tolerance — never compare floats with ==; use small epsilon (0.01) for currency.
Cost — O(N) where N = number of items; each LineItem validation is 5-10 μs; total ~10 μs + 5 μs × N. Negligible for typical orders.

SQL
Topic — SQL
SQL practice library

Practice →

SQL Topic — aggregation SQL aggregation drills

Practice →

3. Pipeline contracts + Kafka payloads

`model_validate_json` at boundaries + DLQ pattern + discriminated unions for schema versioning

The mental model in one line: every ingress boundary in a data pipeline is a chance for malformed data to leak — Kafka topic, HTTP endpoint, CSV file, S3 object, database read of an untrusted source — and Model.model_validate_json(raw_bytes) is the parse-and-validate one-shot that either produces a typed instance or raises ValidationError with structured field-level errors; combine with a dead-letter queue (DLQ) for rejected payloads, a discriminated Literal union for schema versioning, and cross-service contracts backed by JSON Schema and you have a pipeline that fails fast at the boundary instead of silently corrupting downstream aggregates.

Slot 1 — the boundary pattern.

class KafkaEvent(BaseModel):
    id: int
    payload: dict
    ts: datetime

for msg in kafka_consumer:
    try:
        event = KafkaEvent.model_validate_json(msg.value)
    except ValidationError as e:
        dlq.send(msg.value, error=e.errors())
        continue
    process(event)

Parse + validate in one call.
On failure — route to DLQ with structured error.
On success — typed instance for downstream.

Slot 2 — DLQ routing.

What to preserve. Raw payload bytes + structured error + original Kafka offset + timestamp.
Where to route. Separate Kafka topic (orders.dlq) or S3 bucket + notification.
What to alarm. DLQ rate > 1% of ingress rate — page on-call.

Slot 3 — ValidationError.errors() structure.

try:
    KafkaEvent.model_validate_json(b'{"id": "not-int"}')
except ValidationError as e:
    for err in e.errors():
        print(err)
    # {'type': 'int_parsing', 'loc': ('id',), 'msg': 'Input should be a valid integer, ...', 'input': 'not-int'}

Structured, machine-readable — feed into logging / metrics / alerting.

Slot 4 — schema versioning via Literal.

Add a version field to every event.
Literal["v1"] — Pydantic uses this as a discriminator.
Union of versions. Annotated[Union[EventV1, EventV2, EventV3], Field(discriminator="version")].
Dispatch happens at parse time — no runtime isinstance branching in application logic.

Slot 5 — cross-service contracts.

Producer service defines Pydantic model in a shared package (myco-schemas).
Consumer service imports the same package.
Both sides validate. Producer serialises with model_dump_json; consumer parses with model_validate_json.
Alternative — publish JSON Schema to a registry (Confluent Schema Registry, Apicurio); consumers in other languages (Java, Go, TypeScript) generate typed clients.

Slot 6 — FastAPI integration.

from fastapi import FastAPI
app = FastAPI()

class CreateOrderReq(BaseModel):
    customer_id: int
    total: float = Field(ge=0)

@app.post("/orders")
def create_order(req: CreateOrderReq):
    # req is already validated
    return {"id": db.create(req.model_dump())}

FastAPI auto-generates:

422 responses on schema violations.
OpenAPI docs from the model.
/docs interactive UI.

Slot 7 — CSV validation at ETL.

import csv

class OrderRow(BaseModel):
    id: int
    customer_id: int
    total: float
    currency: str = Field(pattern=r"^[A-Z]{3}$")

rejects = []
valid = []
for row in csv.DictReader(open("orders.csv")):
    try:
        valid.append(OrderRow.model_validate(row))
    except ValidationError as e:
        rejects.append({"row": row, "errors": e.errors()})

# Log rejects; bulk-insert valid via SQLAlchemy Core

Slot 8 — the four boundary patterns.

Boundary	Method	Rejection
Kafka	`model_validate_json`	DLQ topic
HTTP	FastAPI + `BaseModel`	422 auto
CSV	`model_validate(dict)`	rejects list
S3 file	`model_validate_json` per line	separate rejects file

Slot 9 — versioning strategies.

Additive fields. New optional field with default — backward compatible; old consumers ignore.
Required field. Version bump. New consumer required.
Removed field. Version bump. Old consumers may fail if they access removed field.
Renamed field. Alias in the new version + deprecation notice.

Slot 10 — cross-service contract patterns.

Shared Python package. Simplest. Both services pip install myco-schemas.
JSON Schema in registry. Language-agnostic. Requires generation step.
Protobuf or Avro. Different tooling; Pydantic can wrap Protobuf via pydantic-protobuf.
GraphQL schema. Different paradigm; Pydantic + Strawberry integrates.

Common beginner mistakes

Not catching ValidationError — worker crashes on malformed input.
Losing the raw payload after rejection — can't debug producer bug.
No DLQ — malformed messages consume retry budget and block the topic.
Trusting versioning by field-name convention — use Literal discriminator.
Not versioning at all — every schema change breaks consumers.
Forgetting to update the schema registry after model change.

Worked example — the DLQ pattern with structured error

Detailed explanation. Consumer processes Kafka messages; malformed messages go to a DLQ with the full error detail for debugging.

Question. Show the complete consumer loop with DLQ routing.

Code.

from pydantic import BaseModel, ValidationError, Field
from datetime import datetime

class OrderEvent(BaseModel):
    id: int
    customer_id: int
    total: float = Field(ge=0)
    ts: datetime

def run_consumer(kafka_consumer, dlq_producer, target):
    processed = 0
    rejected = 0
    for msg in kafka_consumer:
        try:
            event = OrderEvent.model_validate_json(msg.value)
        except ValidationError as e:
            dlq_producer.send(
                topic="orders.dlq",
                key=msg.key,
                value={
                    "original_payload": msg.value.decode("utf-8", errors="replace"),
                    "errors": e.errors(),
                    "original_offset": msg.offset,
                    "original_topic": msg.topic,
                    "rejected_at": datetime.utcnow().isoformat(),
                },
            )
            rejected += 1
            continue
        target.write(event)
        processed += 1
    return processed, rejected

Step-by-step explanation.

Consumer polls Kafka topic.
For each message, try OrderEvent.model_validate_json(msg.value).
On success — event is typed; write to target (DB, downstream Kafka topic).
On ValidationError — build a rich DLQ payload with original bytes, structured error, offset for replay, timestamp.
Send to DLQ topic; continue with next message.
Track processed vs rejected counters; export as metrics.
Alarm if rejected/processed > 1%.

Output.

Metric	Value
Messages consumed	10,000
Processed (valid)	9,950
Rejected (DLQ)	50
Rejection rate	0.5%
Downstream corruption risk	zero

Rule of thumb. Every Kafka consumer needs a DLQ path. Preserve original bytes, structured error, and offset. Alarm on high rejection rate.

Worked example — versioned events via discriminated union

Detailed explanation. Producer emits v1 and v2 of the same event type. Consumer accepts both atomically.

Question. Design the consumer that dispatches v1 vs v2 without isinstance branching in the main loop.

Code.

from typing import Literal, Union, Annotated
from pydantic import BaseModel, Field

class OrderEventV1(BaseModel):
    version: Literal["v1"]
    id: int
    total: float

class OrderEventV2(BaseModel):
    version: Literal["v2"]
    id: int
    total: float
    tenant_id: int
    currency: str = Field(pattern=r"^[A-Z]{3}$")

OrderEvent = Annotated[
    Union[OrderEventV1, OrderEventV2],
    Field(discriminator="version"),
]

class Envelope(BaseModel):
    event: OrderEvent

def process(env: Envelope):
    e = env.event
    match e:
        case OrderEventV1():
            handle_v1(e)
        case OrderEventV2():
            handle_v2(e)

Step-by-step explanation.

Both versions declare version: Literal["v1"|"v2"] — Pydantic uses this literal as the discriminator.
Annotated[Union[...], Field(discriminator="version")] — Pydantic peeks at version, picks the matching model.
Consumer parses via Envelope.model_validate_json(raw) — dispatch happens at parse time.
Downstream match statement branches on concrete type; both branches have full type inference.
Adding v3 later: add OrderEventV3 to union; add match branch; no changes elsewhere.

Output.

Payload	Parsed as	Handler
`{"version":"v1","id":42,"total":100}`	OrderEventV1	handle_v1
`{"version":"v2","id":43,"total":200,"tenant_id":1,"currency":"USD"}`	OrderEventV2	handle_v2
`{"version":"v9",...}`	ValidationError	DLQ

Rule of thumb. Every long-lived event contract should have a version field from day one. Retrofitting versioning is painful.

Worked example — cross-service contract via shared package

Detailed explanation. Producer service and consumer service both pip install myco-schemas for consistent contracts.

Code.

# In shared package myco_schemas/orders.py
from pydantic import BaseModel, Field

class OrderCreated(BaseModel):
    version: str = "v1"
    id: int
    customer_id: int
    total: float = Field(ge=0)

# Producer
event = OrderCreated(id=42, customer_id=1001, total=100.0)
kafka_producer.send("orders", event.model_dump_json().encode())

# Consumer
from myco_schemas.orders import OrderCreated
event = OrderCreated.model_validate_json(msg.value)

Rule of thumb. Shared package is the simplest cross-service contract. For polyglot (Java, Go, TypeScript consumers), publish JSON Schema to a registry.

`pydantic` interview question on ingress-boundary design

A senior interviewer asks: "Design the ingress layer for a webhook receiver that consumes 100K events/day from external partners. Each partner sends slightly different payloads. How do you validate, dispatch, and DLQ?"

Solution Using per-partner discriminated union + shared envelope + DLQ

from pydantic import BaseModel, Field, ValidationError
from typing import Literal, Union, Annotated
from fastapi import FastAPI, HTTPException, Request

class ShopifyOrder(BaseModel):
    source: Literal["shopify"]
    order_id: str
    total_price: str   # Shopify sends as string
    currency: str

class StripeCharge(BaseModel):
    source: Literal["stripe"]
    id: str
    amount: int   # cents
    currency: str

class WebhookEvent(BaseModel):
    event: Annotated[
        Union[ShopifyOrder, StripeCharge],
        Field(discriminator="source"),
    ]

app = FastAPI()

@app.post("/webhook/{partner}")
async def receive(partner: str, request: Request):
    raw = await request.body()
    try:
        # Wrap raw in envelope with source hint
        envelope = WebhookEvent.model_validate({"event": {**parse_partner(partner, raw), "source": partner}})
    except ValidationError as e:
        dlq_send(partner, raw, e.errors())
        raise HTTPException(422, e.errors())

    # Dispatch by concrete type
    e = envelope.event
    match e:
        case ShopifyOrder(): store_shopify(e)
        case StripeCharge(): store_stripe(e)

    return {"ok": True}

Step-by-step trace.

Step	Action
1	Webhook POST arrives with partner query param
2	Read raw body
3	Parse partner-specific format; inject `source` tag
4	Validate via discriminated union
5	On error — DLQ + 422 response
6	On success — match branch dispatches to per-partner handler

Output:

Partner	Model	Payload valid	Response
shopify	ShopifyOrder	yes	200 ok
stripe	StripeCharge	yes	200 ok
shopify (malformed)	—	no	422 + DLQ

Why this works — concept by concept:

Per-partner Pydantic model — each partner's payload has its own model with correct field types (Shopify sends total as string, Stripe as int cents).
Discriminated union — one endpoint handles all partners; dispatch at parse time.
DLQ + 422 — malformed webhooks recorded for debugging; partner gets clear error.
Type-safe match — downstream code branches on concrete type with full inference.
Cost — Pydantic parse ~50 μs; FastAPI overhead ~200 μs; total < 1 ms per webhook, easily handles 100K/day.

SQL
Topic — SQL
SQL practice library

Practice →

SQL Topic — optimization SQL optimization drills

Practice →

4. Pydantic v2 performance + Rust core

`pydantic-core` Rust engine — the 5-50× speedup that made validation affordable in production

The mental model in one line: Pydantic v2's core was rewritten in Rust (the pydantic-core crate) and made available as a compiled Python extension; typical model-parse operations are 5-20× faster than v1, JSON parse-plus-validate is 10-50× faster, and serialization is 3-5× faster — validation is no longer a top-5 CPU item in most pipelines, so the historical excuse "validation is too slow for the hot path" no longer applies, though model_construct remains available for the rare cases where you have already-trusted data.

Slot 1 — the Rust core architecture.

pydantic-core is a separate Rust crate compiled as a Python C extension.
Python pydantic package wraps the core with the developer-friendly API.
Validators are compiled into a schema tree that the Rust core walks per instance.
Zero-copy where possible — string references, JSON parse via serde_json.

Slot 2 — the benchmarks (typical).

Operation	v1 (Python)	v2 (Rust core)	Speedup
Model parse from dict	100 μs	10-20 μs	5-10×
JSON parse + validate	200 μs	5-15 μs	15-40×
Serialize to dict	50 μs	10-20 μs	3-5×
Serialize to JSON	150 μs	20-40 μs	4-7×
Nested model (3 levels)	500 μs	30-60 μs	8-15×

Measured on Python 3.11 with typical small models (< 20 fields).

Slot 3 — when validation cost still matters.

Hot loops processing > 100K records/sec — cumulative overhead can add up.
Very deep nested models — validation cost roughly linear in field count × depth.
Custom validators with heavy logic — your validator function may dominate.
Serialization at high fanout — dumping large lists of models to JSON.

Slot 4 — model_construct — the skip-validation escape hatch.

# Skip validation entirely — for already-trusted data
user = User.model_construct(id=42, name="Alice", email="alice@example.com")

No field-level validation.
No coercion.
No @field_validator execution.
Fastest path — comparable to plain dataclass instantiation.
Use for — rebuilding from trusted store, internal handoffs, unit test fixtures.
Don't use for — external input, anywhere validation would catch a bug.

Slot 5 — model_dump() performance.

v2 default — Python types (datetime, UUID, Decimal) kept as native.
mode="json" — coerce to JSON-safe types (datetime → ISO string, UUID → string).
exclude={"password"} — skip fields.
exclude_unset=True — skip fields that were never explicitly set.
by_alias=True — use field aliases in output.

Slot 6 — string vs dict source performance.

model_validate(dict) — dict already parsed by json.loads; Pydantic walks the dict.
model_validate_json(bytes) — Rust core parses JSON directly; ~1.5-2× faster than manual json.loads + model_validate.
Rule. Always prefer model_validate_json for JSON input.

Slot 7 — nested model overhead.

Flat model (10 fields). ~10-15 μs.
1-level nested (parent + 3 nested children). ~30-50 μs.
2-level nested. ~80-120 μs.
List of 100 nested models. ~1-2 ms.
Rule. Nesting is roughly O(field count × depth).

Slot 8 — validator function overhead.

Simple type check. ~1-2 μs (Rust core inline).
@field_validator with Python function. ~10-30 μs (crosses Python-Rust boundary).
@field_validator with regex. ~5-15 μs (compiled pattern).
@model_validator on 10-field model. ~15-30 μs.

Rule — validator functions are the main overhead. Keep them small and predictable.

Slot 9 — when to profile.

Total pipeline throughput < expected. Profile Python; look for validate_python in hot list.
py-spy top --pid X on running consumer.
scalene script.py for line-level Python + native breakdown.
cProfile for coarser analysis.

Slot 10 — benchmarking Pydantic in your codebase.

import time
from pydantic import BaseModel

class Order(BaseModel):
    id: int
    total: float
    currency: str

payload = b'{"id": 42, "total": 100.0, "currency": "USD"}'

N = 100_000
start = time.perf_counter()
for _ in range(N):
    Order.model_validate_json(payload)
elapsed = time.perf_counter() - start
print(f"{N/elapsed:,.0f} parses/sec, {elapsed*1e6/N:.2f} μs/parse")

Typical output: ~200K parses/sec, ~5 μs/parse on modern hardware.

Common beginner mistakes

Assuming v1 benchmarks still apply to v2 code.
Using model_construct on untrusted data — bypasses all safety.
Not measuring — assuming Pydantic is slow when the bottleneck is DB write.
Forgetting that @field_validator calls cross into Python (expensive) — keep them small.
Not preferring model_validate_json(bytes) over model_validate(json.loads(bytes)).

Worked example — benchmarking v1 vs v2

Detailed explanation. Measure the speedup on the same model between v1 and v2 to quantify the migration win.

Code.

import time
from pydantic import BaseModel

class Order(BaseModel):
    id: int
    customer_id: int
    total: float
    currency: str
    ts: str

payload = b'{"id": 42, "customer_id": 1001, "total": 100.0, "currency": "USD", "ts": "2026-07-12T14:32:00Z"}'

N = 100_000
start = time.perf_counter()
for _ in range(N):
    Order.model_validate_json(payload)
elapsed = time.perf_counter() - start
print(f"v2: {N/elapsed:,.0f} parses/sec ({elapsed*1e6/N:.2f} μs/parse)")

# Typical output:
# v2: 220,000 parses/sec (4.55 μs/parse)
# v1 (same model): 15,000 parses/sec (66.67 μs/parse)
# Speedup: ~14x

Rule of thumb. Migration from v1 to v2 delivers 5-20× speedup with mostly-mechanical code changes. Do it.

Worked example — model_construct for trusted rebuilds

Detailed explanation. After validating a batch, you want to store the raw dicts and later rebuild the models for downstream work. model_construct skips validation for the rebuild.

Code.

# Ingest — full validation
validated: list[Order] = []
for msg in kafka_consumer:
    try:
        order = Order.model_validate_json(msg.value)
    except ValidationError:
        continue
    validated.append(order)
    warehouse.save(order.model_dump())

# Later — rebuild from warehouse without re-validating
def load_batch(rows) -> list[Order]:
    return [Order.model_construct(**row) for row in rows]

Rule of thumb. model_construct for trusted rebuilds is 10-20× faster than model_validate. Only use when data provenance is trusted.

Worked example — profiling a slow pipeline

Detailed explanation. A pipeline claims to process 10K events/sec but actually does 1K. Use profiling to find the bottleneck.

Code.

py-spy top --pid $(pgrep -f consumer)

# Sample output:
%CPU  Function
40%   asyncpg.connection.execute
20%   json.loads
15%   pydantic._internal._model_construction.__init__
10%   consumer.process
5%    kafka poll

Bottleneck is DB, not Pydantic. Optimize the DB write, not the model.

Rule of thumb. Always profile before optimizing. Pydantic v2 is rarely the bottleneck.

`pydantic` interview question on serialization performance

A senior interviewer asks: "Your service serializes 1M Pydantic models to JSON for downstream Kafka. Currently takes 30 seconds. Optimize."

Solution Using model_dump_json + batching + streaming

import orjson
from pydantic import BaseModel

class Order(BaseModel):
    id: int
    total: float

orders: list[Order] = load_orders()   # 1M models

# APPROACH 1: individual dump (slow)
for o in orders:
    kafka.send(o.model_dump_json().encode())
# ~30 seconds

# APPROACH 2: bulk dump with orjson bypass
# model_dump gives dict; orjson dumps to bytes (faster than model_dump_json for bulk)
for o in orders:
    kafka.send(orjson.dumps(o.model_dump()))
# ~12 seconds

# APPROACH 3: bulk buffer + batched send
batch = []
for o in orders:
    batch.append(o.model_dump())
    if len(batch) >= 1000:
        kafka.send_batch(orjson.dumps(batch))
        batch.clear()
# ~5 seconds

# APPROACH 4: parallel workers
# split orders into chunks; process in parallel via ProcessPoolExecutor

Output: 6× speedup via batching + orjson.

Why this works — concept by concept:

orjson for serialization — Rust-based JSON encoder; 2-5× faster than stdlib json and Pydantic's built-in.
Batching — one network round-trip per 1000 models instead of per 1.
model_dump() gives dict — combines with orjson.dumps for the fastest serialization path.
Parallel workers — for CPU-bound serialization on large datasets, spawn workers.
Cost — from 30 μs/model down to ~5 μs/model with batching.

SQL
Topic — optimization
SQL optimization drills

Practice →

SQL Topic — SQL SQL practice library

Practice →

5. Contracts as first-class + dialect matrix

JSON Schema export + OpenAPI + `dataclasses` / `TypedDict` / `SQLModel` — when to use each type-modeling library

The mental model in one line: Pydantic is one of four common Python type-modeling libraries — @dataclass (stdlib, no runtime validation), attrs (like dataclass with more features, no validation), TypedDict (dict shape, type-checker-only), Pydantic BaseModel (full runtime validation) — and SQLModel fuses Pydantic + SQLAlchemy for CRUD apps; the decision is driven by "does this data need runtime validation" (Pydantic yes; others no) and "am I persisting to SQL" (SQLModel maybe); JSON Schema export via Model.model_json_schema() turns Pydantic models into polyglot contracts that Java / Go / TypeScript consumers can generate typed clients from.

Slot 1 — JSON Schema export.

class User(BaseModel):
    id: int
    email: str = Field(pattern=r"^\S+@\S+\.\S+$")

schema = User.model_json_schema()
# {
#   "properties": {
#     "id": {"type": "integer"},
#     "email": {"type": "string", "pattern": "^\\S+@\\S+\\.\\S+$"}
#   },
#   "required": ["id", "email"],
#   "title": "User",
#   "type": "object"
# }

Publish to schema registry; consumers in any language generate typed clients.

Slot 2 — OpenAPI via FastAPI.

FastAPI auto-generates OpenAPI 3 spec from Pydantic models used in route signatures:

from fastapi import FastAPI
app = FastAPI()

@app.post("/orders")
def create(order: Order) -> OrderResponse:
    ...

Visit /openapi.json for the spec, /docs for interactive UI.

Slot 3 — @dataclass.

from dataclasses import dataclass

@dataclass(frozen=True, slots=True)
class ProcessedOrder:
    id: int
    total: float

No runtime validation.
Type hints only (mypy / pyright checks).
Fastest instantiation (~1 μs).
Best for internal trusted state.

Slot 4 — attrs.

import attrs

@attrs.frozen
class Config:
    host: str
    port: int = 5432

    @port.validator
    def _check_port(self, attribute, value):
        if not (1 <= value <= 65535):
            raise ValueError("port out of range")

Like dataclass, but with per-field validators.
Older than dataclass; still widely used.
No runtime type coercion by default.

Slot 5 — TypedDict.

from typing import TypedDict

class UserDict(TypedDict):
    id: int
    name: str
    email: str

def process(user: UserDict) -> None:
    print(user["name"])

Type-checker hint on dict shape.
Zero runtime cost.
Best for legacy dict-based code you want to type without refactoring.

Slot 6 — SQLModel.

from sqlmodel import SQLModel, Field

class User(SQLModel, table=True):
    id: int | None = Field(default=None, primary_key=True)
    name: str
    email: str

Pydantic + SQLAlchemy fusion.
One class for API contract + DB schema.
Good for CRUD apps.
Not great for complex DB layouts (raw SQLAlchemy Core wins).

Slot 7 — the 5-library decision matrix.

Library	Runtime validation	Type-check	Persistence	Best for
`@dataclass`	No	Yes	No	Internal state
`attrs`	Optional (validator)	Yes	No	Internal state with checks
`TypedDict`	No	Yes	No	Legacy dict typing
Pydantic `BaseModel`	Yes	Yes	No	Ingress boundaries
`SQLModel`	Yes	Yes	Yes (SQLAlchemy)	CRUD apps

Slot 8 — serialization options.

model_dump() — → dict, Python types kept.
model_dump(mode="json") — → dict, JSON-safe types (datetime → str).
model_dump_json() — → JSON string (Pydantic built-in).
orjson.dumps(m.model_dump()) — → JSON bytes (faster for bulk).
model_dump(exclude={"password"}) — omit fields.
model_dump(exclude_unset=True) — omit fields never explicitly set.
model_dump(by_alias=True) — use field aliases.

Slot 9 — pydantic-settings for config.

from pydantic_settings import BaseSettings, SettingsConfigDict

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env", env_prefix="APP_")
    database_url: str
    kafka_brokers: str
    log_level: str = "INFO"

settings = Settings()   # loads from env / .env

Typed config from env vars.
App refuses to start if required config missing.
Validates via same Pydantic mechanics.

Slot 10 — the 8-integration matrix.

Framework	Pydantic role
FastAPI	request/response models + OpenAPI
LangChain	structured output
SQLModel	ORM + API
Dagster	asset materialization types
Prefect	flow parameters
MLflow	model signatures
dbt	(via python models)
DuckDB	(indirect via arrow)

Pydantic is the de-facto type surface across modern Python data stack.

Common beginner mistakes

Using Pydantic for internal state where dataclass suffices.
Using dataclass for ingress where you need runtime validation.
Not publishing JSON Schema for cross-service contracts.
Trying to use Pydantic as an ORM (use SQLModel or SQLAlchemy).
Mixing v1 .dict() and v2 .model_dump() in same codebase.

Worked example — publish JSON Schema to a registry

Code.

import json
from pydantic import BaseModel, Field

class OrderCreated(BaseModel):
    version: str = "v1"
    id: int
    total: float = Field(ge=0)

schema = OrderCreated.model_json_schema()
# Publish
with open("schemas/orders/OrderCreated-v1.json", "w") as f:
    json.dump(schema, f, indent=2)

# Or push to Confluent Schema Registry
registry.register("orders-value", json.dumps(schema))

Consumers in Java / Go / TypeScript can generate typed clients from the schema.

Rule of thumb. Every cross-service Pydantic model deserves a JSON Schema in the registry.

Worked example — `pydantic-settings` for typed config

Code.

from pydantic import Field
from pydantic_settings import BaseSettings, SettingsConfigDict

class DatabaseSettings(BaseSettings):
    model_config = SettingsConfigDict(env_prefix="DB_")
    url: str
    pool_size: int = Field(default=5, ge=1, le=50)
    ssl: bool = True

class Settings(BaseSettings):
    model_config = SettingsConfigDict(env_file=".env")
    database: DatabaseSettings = Field(default_factory=DatabaseSettings)
    log_level: str = "INFO"
    kafka_brokers: list[str]   # comma-separated in env

settings = Settings()
# App fails to start if DB_URL missing.

Rule of thumb. Typed config catches deployment errors at startup, not at runtime.

Worked example — dataclass vs BaseModel benchmark

Code.

import time
from dataclasses import dataclass
from pydantic import BaseModel

@dataclass
class OrderDC:
    id: int
    total: float

class OrderBM(BaseModel):
    id: int
    total: float

N = 1_000_000

start = time.perf_counter()
for _ in range(N):
    OrderDC(id=42, total=100.0)
print(f"dataclass: {N/(time.perf_counter()-start):,.0f}/sec")

start = time.perf_counter()
for _ in range(N):
    OrderBM(id=42, total=100.0)
print(f"BaseModel: {N/(time.perf_counter()-start):,.0f}/sec")

# Typical output:
# dataclass: 5,000,000/sec
# BaseModel: 500,000/sec

Rule of thumb. dataclass is ~10× faster than BaseModel for pure instantiation. Use BaseModel where validation matters; dataclass everywhere else.

`pydantic` interview question on contract publication

A senior interviewer asks: "Design the schema publication process for a data platform serving 20 teams. Producers in Python; consumers in Python, Java, Go. How do you keep contracts in sync?"

Solution Using shared repo + Pydantic authored + CI-generated JSON Schema + polyglot clients

myco-schemas/
├── python/                    # authored here
│   ├── orders/v1.py           # Pydantic BaseModel
│   ├── users/v1.py
│   └── payments/v2.py
├── schemas/                   # CI-generated
│   ├── orders-v1.json
│   ├── users-v1.json
│   └── payments-v2.json
├── generated/                 # CI-generated
│   ├── java/                  # from JSON Schema via quicktype/openapi-generator
│   ├── go/
│   └── typescript/
└── ci/
    └── validate-and-generate.yml

CI pipeline.

Author Pydantic model in python/orders/v1.py.
CI runs model_json_schema() and writes to schemas/.
CI runs quicktype (or similar) to generate Java / Go / TypeScript typed clients.
CI publishes each language client to its respective package registry (Maven, Go modules, npm).
Consumers pin exact version.

Why this works — concept by concept:

Pydantic as source of truth — single language for schema authorship.
JSON Schema as the wire contract — polyglot; every language has JSON Schema tools.
CI generates clients — no manual sync between languages.
Semantic versioning — breaking changes bump major version; add-only bumps minor.
Schema registry — optional additional step for Kafka runtime validation.

SQL
Topic — SQL
SQL practice library

Practice →

SQL
Topic — optimization
SQL optimization drills

Practice →

Cheat sheet — Pydantic recipe list

from pydantic import BaseModel, Field, ValidationError, field_validator, model_validator, ConfigDict — the core imports.
class X(BaseModel): — inherit; every annotated field becomes runtime-validated.
Field(gt=0, ge=0, lt=100, le=100) — numeric constraints.
Field(min_length=1, max_length=100, pattern=r"...") — string constraints.
Field(default_factory=list) — computed default per instance.
Field(alias="userId") — accept alternate wire name.
Optional[T] = None or T | None = None — nullable + optional.
@field_validator("col", mode="before") — per-field, before coercion.
@field_validator("col", mode="after") — per-field, after coercion.
@model_validator(mode="after") — cross-field business rules.
Model.model_validate(dict) — dict → model.
Model.model_validate_json(bytes) — bytes → model in one call (fastest).
model.model_dump() — model → dict, Python types.
model.model_dump(mode="json") — → dict, JSON-safe types.
model.model_dump_json() — → JSON string.
model.model_construct(...) — skip validation (trusted paths only).
ConfigDict(strict=True) — no coercion.
ConfigDict(str_strip_whitespace=True) — auto-trim.
ConfigDict(frozen=True) — immutable instance.
ConfigDict(extra="forbid") — reject unknown fields.
ConfigDict(populate_by_name=True) — accept both alias and name.
ConfigDict(from_attributes=True) — parse from any object with matching attrs.
Literal["v1"] + discriminated union — schema versioning.
Annotated[Union[V1,V2], Field(discriminator="version")] — dispatch at parse.
EmailStr — email with validation (pip install pydantic[email]).
HttpUrl — URL with validation.
Decimal — for money; better than float.
datetime fields — parse ISO 8601 automatically.
UUID — parse UUID strings.
e.errors() — structured error list on ValidationError.
DLQ pattern — catch ValidationError, preserve payload + error + offset.
model_json_schema() — export JSON Schema for cross-service contracts.
FastAPI + Pydantic — automatic 422 responses + OpenAPI docs.
pydantic-settings — typed config from env vars.
SQLModel — Pydantic + SQLAlchemy fusion for CRUD.
orjson.dumps(model.model_dump()) — fastest JSON output for bulk.
v2 Rust core — 5-50× faster than v1; measure before optimizing.
v2 migration — mechanical rename .dict() → .model_dump(), @validator → @field_validator.

Frequently asked questions

When should I use Pydantic instead of `@dataclass`?

Use Pydantic when your data comes from an untrusted source and needs runtime validation — Kafka payloads, HTTP requests, CSV rows, config files, external API responses. Use @dataclass for internal trusted state where you want the ergonomic benefits (typed attributes, auto __init__, frozen=True) without the ~10× cost of runtime validation. Common pattern: Pydantic at ingress boundary, dataclass everywhere internal, model_construct when rebuilding Pydantic models from already-trusted data. Both can coexist in the same codebase; convert between them explicitly.

What's the difference between `mode="before"` and `mode="after"` on `@field_validator`?

mode="before" runs BEFORE Pydantic's type coercion — the validator receives the raw input (usually a string or dict) and can transform it into something Pydantic then coerces. Use for currency-symbol stripping, whitespace normalization, custom parsing. mode="after" (default) runs AFTER type coercion — the validator receives the coerced value (int, float, datetime) and can enforce further constraints. Use for domain-specific validation that requires the parsed value. If you need both, define two validators.

How do I handle schema versioning?

Add a version field with type Literal["v1"] to every event/payload model. Build a discriminated union: Annotated[Union[EventV1, EventV2], Field(discriminator="version")]. Pydantic dispatches to the right model at parse time based on the version tag; consumer code branches on the concrete type via match or isinstance. Adding a new version means defining EventV3, adding it to the union, and adding a match branch — no changes to existing paths. Every event contract should carry a version tag from day one; retrofitting is painful.

Is Pydantic v2 really 10× faster than v1?

Yes, for parse-and-validate operations, typically 5-20× on model parse, 10-50× on JSON parse-plus-validate, 3-5× on serialization. The core was rewritten in Rust as pydantic-core; the Python pydantic package is a thin API wrapper. Micro-benchmark on your models before migrating (typical output: ~200K parses/sec on v2 vs ~15K/sec on v1). Migration from v1 to v2 is mostly mechanical: .dict() → .model_dump(), @validator → @field_validator, Config inner class → model_config = ConfigDict(...).

When should I use `model_construct`?

model_construct skips all validation, coercion, and validator execution — it's the escape hatch for trusted internal handoffs where the data has already been validated upstream. Common uses: rebuilding a Pydantic model from a warehouse row that was written by the same pipeline (data provenance is trusted), unit test fixtures where you know the values are correct, internal hot loops where validation cost dominates. Never use on external input — you lose all the safety guarantees. Typical speedup vs model_validate: 10-20×.

How do I integrate Pydantic with SQLAlchemy?

Two options. Option 1 — separate models: use SQLAlchemy Mapped classes for persistence and Pydantic BaseModel for API/wire contracts; convert between them explicitly with Pydantic.from_orm(sqla_obj) (v1) or Pydantic.model_validate(sqla_obj, from_attributes=True) (v2). Clean separation of concerns. Option 2 — SQLModel: a Pydantic + SQLAlchemy fusion library. One class serves as both ORM and API model. Good for CRUD-heavy apps; less flexible for complex DB layouts where raw SQLAlchemy Core wins. Rule: SQLModel for straightforward CRUD; separate models for complex data platforms.

Practice on PipeCode

Drill the SQL practice library → — 450+ DE-focused questions covering pipeline contracts and validation.
Sharpen SQL optimization drills → for reading validated ingress + downstream aggregate correctness.
Layer SQL aggregation drills → — clean ingress feeds correct aggregates.
Warm up with SQL join drills → — pipeline correctness relies on typed join keys.
For the broader SQL interview surface, take the SQL for Data Engineering course →.

Pipecode.ai is Leetcode for Data Engineering — every `pydantic` pattern above ships with hands-on practice rooms where you build the boundary validator with `BaseModel` + `Field(gt=0)` + `@field_validator`, wire the DLQ path on `ValidationError`, migrate v1 to v2 with the mechanical rename table, benchmark the Rust core against v1 to see the 10× speedup, publish JSON Schema for cross-service polyglot contracts, and finally choose between Pydantic, dataclass, TypedDict, attrs, and SQLModel per workload — the exact Python-side contract fluency that senior DE interviews probe. PipeCode pairs every Pydantic concept with 450+ DE-focused problems and a real-time scoring engine, so you never have to wonder whether your `BaseModel` / `model_validate_json` / discriminated-union answer holds up under a senior interviewer's depth probes.

Practice SQL now →
Optimization drills →

1. Why Pydantic matters for DE in 2026

The pydantic mental model — validate once at the ingress boundary, trust the typed model through the rest of the pipeline

Worked example — the silent NoneType bug that Pydantic catches

Worked example — the type-coercion bug on CSV int fields

Worked example — the contract-drift bug and schema versioning

pydantic interview question on choosing between raw dict, dataclass, and BaseModel

Solution Using ingress-boundary Pydantic + internal dataclass hand-off + model_construct escape hatch

2. BaseModel + Field validators

pydantic BaseModel + Field constraints + @field_validator + @model_validator — the four primitives that cover 90% of DE validation

Worked example — a complete Order model with all four primitives

Worked example — refusing coercion with strict mode

Worked example — model_validate_json vs model_validate

pydantic BaseModel interview question on nested models and lists

Solution Using nested BaseModel + list[Item] + model_validator for total invariant

3. Pipeline contracts + Kafka payloads

model_validate_json at boundaries + DLQ pattern + discriminated unions for schema versioning

Worked example — the DLQ pattern with structured error

Worked example — versioned events via discriminated union

Worked example — cross-service contract via shared package

pydantic interview question on ingress-boundary design

Solution Using per-partner discriminated union + shared envelope + DLQ

4. Pydantic v2 performance + Rust core

pydantic-core Rust engine — the 5-50× speedup that made validation affordable in production

Worked example — benchmarking v1 vs v2

Worked example — model_construct for trusted rebuilds

Worked example — profiling a slow pipeline

pydantic interview question on serialization performance

Solution Using model_dump_json + batching + streaming

5. Contracts as first-class + dialect matrix

JSON Schema export + OpenAPI + dataclasses / TypedDict / SQLModel — when to use each type-modeling library

Worked example — publish JSON Schema to a registry

Worked example — pydantic-settings for typed config

Worked example — dataclass vs BaseModel benchmark

pydantic interview question on contract publication

Solution Using shared repo + Pydantic authored + CI-generated JSON Schema + polyglot clients

Cheat sheet — Pydantic recipe list

Frequently asked questions

When should I use Pydantic instead of @dataclass?

What's the difference between mode="before" and mode="after" on @field_validator?

How do I handle schema versioning?

Is Pydantic v2 really 10× faster than v1?

When should I use model_construct?

How do I integrate Pydantic with SQLAlchemy?

Practice on PipeCode

The `pydantic` mental model — validate once at the ingress boundary, trust the typed model through the rest of the pipeline

`pydantic` interview question on choosing between raw dict, dataclass, and BaseModel

`pydantic BaseModel` + `Field` constraints + `@field_validator` + `@model_validator` — the four primitives that cover 90% of DE validation

Worked example — `model_validate_json` vs `model_validate`

`pydantic BaseModel` interview question on nested models and lists

`model_validate_json` at boundaries + DLQ pattern + discriminated unions for schema versioning

`pydantic` interview question on ingress-boundary design

`pydantic-core` Rust engine — the 5-50× speedup that made validation affordable in production

`pydantic` interview question on serialization performance

JSON Schema export + OpenAPI + `dataclasses` / `TypedDict` / `SQLModel` — when to use each type-modeling library

Worked example — `pydantic-settings` for typed config

`pydantic` interview question on contract publication

When should I use Pydantic instead of `@dataclass`?

What's the difference between `mode="before"` and `mode="after"` on `@field_validator`?

When should I use `model_construct`?