Schema Contracts and Row Validation

25.1 Why Contracts? (20 min)

Recall Day 24.3: the field geo was renamed, dlt happily evolved, and SQL filtering on geo = 'Canada' silently missed every new row. Nothing failed. The dashboard just started showing stale numbers.

A data contract states what a table is allowed to look like. When incoming data breaks it, the pipeline reacts in a way you chose: stop, drop the row, or drop the value.

There are three things a contract can govern:

25.2 Contract Modes (30 min)

Each entity can take one of four modes:

Create day25/contracts.py. Each experiment starts from the same baseline table:

from common import practice_pipeline, reset

HINTS = {"value": {"data_type": "double"}}
BASE = [{"series_id": "V39079", "value": 2.25}]


def fresh():
    reset("contract_lab")
    p = practice_pipeline("contract_lab")
    p.run(BASE, table_name="obs", columns=HINTS)          # baseline: 2 columns
    return p


def show(p):
    import duckdb
    con = duckdb.connect("contract_lab.duckdb", read_only=True)
    print(con.sql("SELECT * EXCLUDE (_dlt_load_id, _dlt_id) FROM lab.obs").fetchall())
    print(con.sql("SELECT column_name FROM information_schema.columns "
                  "WHERE table_name = 'obs' AND column_name NOT LIKE '\\_dlt%' ESCAPE '\\'").fetchall())
    con.close()

Experiment 1 — freeze columns

p = fresh()
try:
    p.run([{"series_id": "V39079", "value": 2.50, "status": "E"}],
          table_name="obs", schema_contract={"columns": "freeze"})
except Exception as e:
    print("Blocked:", type(e).__name__)
show(p)

The new status column is rejected and nothing from that load is written.

Experiment 2 — discard_value columns

p = fresh()
p.run([{"series_id": "V39079", "value": 2.50, "status": "E"}],
      table_name="obs", schema_contract={"columns": "discard_value"})
show(p)

The row loads and the unknown status value is silently dropped.

Experiment 3 — discard_row columns

p = fresh()
p.run([{"series_id": "V39079", "value": 2.50, "status": "E"},
       {"series_id": "V39079", "value": 2.75}],
      table_name="obs", schema_contract={"columns": "discard_row"})
show(p)

Only the row without the unexpected field loads.

Experiment 4 — freeze data types

p = fresh()
try:
    p.run([{"series_id": "V39079", "value": "NA"}],
          table_name="obs", schema_contract={"data_type": "freeze"})
except Exception as e:
    print("Blocked:", type(e).__name__)

No variant column is created; the load fails instead.

Shorthand

A single string applies to all three entities:

schema_contract="freeze"          # same as {"tables": "freeze", "columns": "freeze", "data_type": "freeze"}

New tables are exempt from column rules. The first load of a table defines its columns, so contracts on columns and data_type protect tables that already exist. tables: "freeze" stops new tables from being created at all.

Try it 25.2

Which mode would you choose, and why, for: (a) a raw landing table of Valet responses; (b) the boc_observations table feeding your SQL staging models; (c) a table of user-submitted survey answers where an occasional malformed row is expected?

25.3 Where to Set Contracts (20 min)

Contracts can be set at three levels. More specific settings override more general ones:

# 1. On a resource — applies to that table
@dlt.resource(name="boc_observations",
              schema_contract={"columns": "freeze", "data_type": "freeze"})
def observations(): ...

# 2. On a source — default for all its resources
@dlt.source(name="boc", schema_contract={"tables": "evolve", "columns": "freeze"})
def boc_source(): ...

# 3. On a single run — overrides for this run only
pipeline.run(source, schema_contract="evolve")

A layered strategy

raw_json / landing tables     →  evolve        (capture everything)
raw_dlt observation tables    →  freeze        (shape is a promise to SQL models)
metadata tables               →  columns: evolve, data_type: freeze
                                  (new descriptive fields are welcome; type changes are not)

Controlled evolution: when you intend a change (say, adding a unit column), run once with schema_contract="evolve" on the command line or a flag, review the exported schema diff (Day 24.6), commit, and return to freeze. Changing the schema is a deliberate act, not an accident.

25.4 Handling Violations Gracefully (20 min)

A frozen contract raises an exception during the normalize step. The pipeline wraps it in a PipelineStepFailed error, which tells you which step failed. Your runner should:

1. catch it,
2. log clearly which source and what changed,
3. keep other sources running (Day 9's design),
4. exit with a non-zero code so schedulers notice (Day 10).

from dlt.pipeline.exceptions import PipelineStepFailed


def run_with_contract(pipeline, source, name):
    try:
        info = pipeline.run(source)
        info.raise_on_failed_jobs()
        print(f"✓ {name}")
        return True
    except PipelineStepFailed as e:
        print(f"✗ {name}: failed at step '{e.step}'")
        cause = e.__context__ or e
        print(f"   cause: {type(cause).__name__}: {str(cause)[:300]}")
        return False

The underlying error (often a DataValidationError) names the table, column and contract mode involved, so you know immediately what the API changed.

What happens to data that failed?

The extracted data from a failed normalize step stays in the pipeline's working folder as a pending package. After fixing the contract (or deciding to evolve), the next pipeline.run() can process it. For simplicity in this course, prefer re-running the source after a fix. Pending packages are covered in dlt's documentation if you need them.

25.5 Row Validation with Pydantic Models (25 min)

Contracts check the shape of data. Pydantic models add checks on values, written as a typed Python class:

pip install pydantic
pip freeze > requirements.txt

Create day25/pydantic_models.py:

from datetime import date
from typing import Literal

from pydantic import BaseModel, Field, field_validator


class Observation(BaseModel):
    """One observation of one macro series (our long-format contract)."""
    date: date
    series_id: str = Field(min_length=2)
    value: float
    source: Literal["boc", "fred", "statcan"]

    @field_validator("value")
    @classmethod
    def must_be_finite(cls, v):
        if v != v or v in (float("inf"), float("-inf")):     # NaN is never equal to itself
            raise ValueError("value must be a finite number")
        return v


# Try it without dlt first
print(Observation(date="2026-07-01", series_id="V39079", value="2.25", source="boc"))
try:
    Observation(date="2026-07-01", series_id="V39079", value=2.25, source="imf")
except Exception as e:
    print(e)

Pydantic converts where it's safe ("2.25" → 2.25, "2026-07-01" → a date) and rejects what doesn't fit (an unknown source).

Using a model as a dlt resource's columns

import dlt


@dlt.resource(name="validated_obs", columns=Observation,
              schema_contract={"columns": "discard_value", "data_type": "discard_row"})
def validated_obs():
    yield [
        {"date": "2026-07-01", "series_id": "V39079", "value": 2.25, "source": "boc"},
        {"date": "2026-07-02", "series_id": "V39079", "value": "oops", "source": "boc"},
        {"date": "2026-07-03", "series_id": "V39079", "value": 2.25, "source": "boc",
         "extra": "ignored"},
    ]

With a Pydantic model as columns, dlt:

• derives the table's column types from the model (no separate hints needed);
• validates each item against the model, applying the contract modes when items don't match. Here the row with "oops" is discarded and the unexpected extra field is dropped.

Run it into a practice pipeline and check which rows arrived.

Where do economic plausibility checks belong? "CPI must be positive" or "unemployment between 0 and 100" can live in Pydantic, but it's usually better to load the raw values and test them downstream (Day 18's SQL tests, Great Expectations on Days 41–50). Rejecting a surprising value at ingestion hides it; testing it downstream shows it to you. Keep ingestion validation structural (types, required fields, allowed codes).

25.6 Mini-Project: Contracts for macro_ingest (20 min)

1. Add schema_contract={"columns": "freeze", "data_type": "freeze"} to the three observation resources in macro_ingest, and {"columns": "evolve", "data_type": "freeze"} to the three metadata resources.
2. Update ingest.py to use run_with_contract (25.4).
3. Add export_schema_path="schemas/export" to the macro_ingest pipeline and commit the exported schema.
4. Write day25/contract_test.py that proves the contract works: run the BoC source with add_map injecting an extra field into every row, into a throwaway DuckDB file, and assert that the load fails.

"""Prove that the frozen contract on boc_observations blocks unexpected columns."""
import os

import dlt
from dlt.pipeline.exceptions import PipelineStepFailed

from macro_ingest.boc import boc_source
from common import reset

reset("contract_test")
pipeline = dlt.pipeline(pipeline_name="contract_test",
                        destination=dlt.destinations.duckdb("contract_test.duckdb"),
                        dataset_name="test")

# 1. Baseline load establishes the table (contracts protect existing tables)
pipeline.run(boc_source(series_ids=["V39079"], start_date="2026-01-01")
             .with_resources("boc_observations"))

# 2. Simulate an API change: every row gains a new field
def add_field(row):
    row["surprise"] = "new field"
    return row

source = boc_source(series_ids=["V39079"], start_date="2026-01-01")
source.resources["boc_observations"].add_map(add_field)

try:
    pipeline.run(source.with_resources("boc_observations"))
    raise AssertionError("Contract did NOT block the new column!")
except PipelineStepFailed:
    print("✓ Contract blocked the unexpected column, as intended")
finally:
    reset("contract_test")

Day 25 checklist

• I can explain the four contract modes and the three entities they govern
• I choose contracts per layer: evolve for raw capture, freeze for tables that SQL models depend on
• My runner catches contract violations, reports them clearly and exits non-zero
• I can validate rows with a Pydantic model and use it as a resource's columns
• I keep ingestion checks structural and leave economic plausibility to downstream tests
• I have a test that proves my contract works

git add .
git commit -m "Day 25: schema contracts, Pydantic validation, contract test"
git push