Project: Kiosko's first data contract

Description

This project closes the module. You have the precise definition of a contract (lesson 2), S04's real file written section by section (lesson 3), the parsing with pydantic and the structural guarantee it brings (lesson 4), the function that generates an executable schema from the contract (lesson 5), the versioning discipline for when the contract changes (lesson 6), and the governance question about admission (lesson 7). One step remains: a closing script, end to end, that parses the real contract, generates the schema, runs it against S04, and confirms with evidence the module's complete promise.

Connection to the module. This project introduces no new concept — it's the complete synthesis of lessons 2 through 7, applied end to end to the same incident that modules 1, 2, and 3 diagnosed. It closes this module's thread exactly where module 5 picks it back up: ORD-9509, the row with the 60.00 price, is still waiting — no contract in this module has any way to express "compare this value against a historical baseline."

An analogy: the lease signed, sealed, and filed away — not just drafted

A lease that exists only as a draft on the landlord's computer doesn't protect anyone yet — it needs to be signed, handed to both parties, and kept on file somewhere either party can consult it if needed. This project is that final step: orders_contract.yaml stops being a file that earlier lessons built piece by piece, and becomes the artifact a real script uses, from start to finish, to make a decision about real data.

The material you need

You need, in this module's same working folder:

module_4_contracts/
├── kiosko.duckdb              (orders_s04 from module 2)
├── orders_contract.yaml       (lesson 3)
└── project_contract.py        (this project)

If your kiosko.duckdb doesn't have orders_s04 yet, repeat step 2 of module 2's lesson 4 before continuing — this project doesn't re-explain that step, it assumes you've already done it.

The verified reference solution

# project_contract.py -- module 4 closing project
from typing import Literal

import duckdb
import pandera
import pandera.polars as pa
import polars as pl
import yaml
from pydantic import BaseModel, Field


class ColumnContract(BaseModel):
    name: str
    type: Literal["string", "float", "integer"]
    nullable: bool = True
    unique: bool = False
    minimum: float | None = None
    exclusive_minimum: float | None = None


class RowCountRange(BaseModel):
    min: int
    max: int


class SLAContract(BaseModel):
    freshness_hours: int
    row_count: RowCountRange


class DataContract(BaseModel):
    contract_version: str
    dataset: str
    owner: str
    description: str
    schema_: list[ColumnContract] = Field(alias="schema")
    sla: SLAContract
    on_violation: Literal["quarantine", "reject", "alert"]


TYPE_MAP = {"string": str, "float": float, "integer": int}


def contract_to_pandera_schema(contract: DataContract) -> pa.DataFrameSchema:
    """Convert an already-parsed DataContract into an executable pa.DataFrameSchema."""
    columns = {}
    for col in contract.schema_:
        cast = int if col.type == "integer" else float
        checks = []
        if col.minimum is not None:
            checks.append(pa.Check.ge(cast(col.minimum)))
        if col.exclusive_minimum is not None:
            checks.append(pa.Check.gt(cast(col.exclusive_minimum)))
        columns[col.name] = pa.Column(
            TYPE_MAP[col.type], checks=checks, nullable=col.nullable, unique=col.unique
        )
    return pa.DataFrameSchema(columns)


CHECK_TO_DIMENSION = {
    "not_nullable": "completeness",
    "field_uniqueness": "uniqueness",
    "greater_than(0)": "validity",
}


def main() -> None:
    print("=== Kiosko: the first data contract, end to end ===")
    print(f"pandera version: {pandera.__version__}\n")

    with open("orders_contract.yaml") as f:
        raw = yaml.safe_load(f)
    contract = DataContract.model_validate(raw)
    print(f"Contract parsed: {contract.dataset} v{contract.contract_version} (owner={contract.owner})")
    print(f"SLA: freshness_hours={contract.sla.freshness_hours}, "
          f"row_count=[{contract.sla.row_count.min}, {contract.sla.row_count.max}]")
    print(f"on_violation: {contract.on_violation}\n")

    generated_schema = contract_to_pandera_schema(contract)

    con = duckdb.connect("kiosko.duckdb")
    df = con.sql("SELECT * FROM orders_s04").pl()
    print(f"Rows read from orders_s04: {df.height}")

    row_count_ok = contract.sla.row_count.min <= df.height <= contract.sla.row_count.max
    print(f"contract's implicit check_volume: {df.height} rows, "
          f"range [{contract.sla.row_count.min}, {contract.sla.row_count.max}] -> "
          f"{'PASSES' if row_count_ok else 'FAILS'}\n")

    try:
        generated_schema.validate(df, lazy=True)
        print("All rows passed the generated schema (not expected).")
        return
    except pa.errors.SchemaErrors as exc:
        fc = exc.failure_cases

    readable = fc.with_columns(
        pl.col("index").map_elements(lambda i: df["order_id"][i], return_dtype=pl.String).alias("order_id")
    ).select(["order_id", "column", "check", "failure_case"])

    print("=== contract_to_pandera_schema(contract).validate(df, lazy=True): SchemaErrors ===")
    print(readable)

    failing_indices = sorted(fc["index"].unique().to_list())
    dimensions = sorted({CHECK_TO_DIMENSION[c] for c in fc["check"].to_list()})

    print(f"\nPhysical rows with at least one error: {len(failing_indices)} of {df.height}")
    print(f"Quality dimensions caught: {len(dimensions)} of 6 ({dimensions})")

    print("\n=== Comparison against module 2 (hand-written OrdersSchema) ===")
    print("Module 2, OrdersSchema (by hand): 4 rows with an error, 3 dimensions (completeness, uniqueness, validity)")
    print(f"Module 4, contract_to_pandera_schema(contract) (generated): "
          f"{len(failing_indices)} rows with an error, {len(dimensions)} dimensions {dimensions}")
    print("Exact same result -- the contract GENERATES the test, the test no longer lives separately from the contract.")


if __name__ == "__main__":
    main()

What to expect (verified by actually running python3 project_contract.py, with kiosko.duckdb and orders_contract.yaml in the same folder, pandera==0.32.1):

=== Kiosko: the first data contract, end to end ===
pandera version: 0.32.1

Contract parsed: orders_s04 v1.0.0 (owner=kiosko-data-team)
SLA: freshness_hours=24, row_count=[5, 20]
on_violation: quarantine

Rows read from orders_s04: 12
contract's implicit check_volume: 12 rows, range [5, 20] -> PASSES

=== contract_to_pandera_schema(contract).validate(df, lazy=True): SchemaErrors ===
shape: (4, 4)
┌──────────┬────────────┬──────────────────┬──────────────┐
│ order_id ┆ column     ┆ check            ┆ failure_case │
│ ---      ┆ ---        ┆ ---              ┆ ---          │
│ str      ┆ str        ┆ str              ┆ str          │
╞══════════╪════════════╪══════════════════╪══════════════╡
│ ORD-9502 ┆ order_id   ┆ field_uniqueness ┆ ORD-9502     │
│ ORD-9502 ┆ order_id   ┆ field_uniqueness ┆ ORD-9502     │
│ ORD-9503 ┆ unit_price ┆ not_nullable     ┆ null         │
│ ORD-9507 ┆ quantity   ┆ greater_than(0)  ┆ -1           │
└──────────┴────────────┴──────────────────┴──────────────┘

Physical rows with at least one error: 4 of 12
Quality dimensions caught: 3 of 6 (['completeness', 'uniqueness', 'validity'])

=== Comparison against module 2 (hand-written OrdersSchema) ===
Module 2, OrdersSchema (by hand): 4 rows with an error, 3 dimensions (completeness, uniqueness, validity)
Module 4, contract_to_pandera_schema(contract) (generated): 4 rows with an error, 3 dimensions ['completeness', 'uniqueness', 'validity']
Exact same result -- the contract GENERATES the test, the test no longer lives separately from the contract.

Read the full result with the same care you've already trained in earlier projects. The metadata block confirms the contract got read, parsed, and structurally validated with no error at all — version, owner, SLA, policy, all accessible as typed attributes. contract's implicit check_volume is an extra confirmation this project adds, needing no new tool at all: twelve rows, within the [5, 20] range the SLA declares — an honest preview of what module 6 is going to formalize with its own function, check_volume(). And the central result, already known but now derived from a versioned artifact instead of hand-written code: four physical rows with an error, three dimensions — exactly the same as module 2 already confirmed.

Diagram: where you came from, where you arrived

flowchart LR
    A["Module 2:\nOrdersSchema written\nby hand in Python"] --> B["Lesson 2:\na complete contract =\nschema + SLA + policy"]
    B --> C["Lesson 3:\norders_contract.yaml\nwritten, section by section"]
    C --> D["Lesson 4:\nDataContract/ColumnContract\n(pydantic), broken YAML tested"]
    D --> E["Lesson 5:\ncontract_to_pandera_schema()\nsame result as M2"]
    E --> F["Lesson 6:\nv1.1.0 compatible,\nv2.0.0 breaks compatibility"]
    F --> G["Lesson 7:\nadmission_check() --\nS04 would have been blocked"]
    G --> H["This project:\nend-to-end contract,\n4 rows with an error, 3 dimensions"]
    H --> I["Module 5:\naccuracy -- ORD-9509\nstill uncaught"]

Closing the module's promise, point by point

With this project, orders_contract.yaml stops being one module's isolated file — from here on, it's the artifact any future validation of S04 in this guide should start from. Module 5 doesn't rewrite any schema from scratch: it builds a completely new tool — a price baseline — because accuracy is, precisely, the kind of rule no declarative schema, whether hand-written or generated from a contract, can express.

Common mistakes

Considering S04's whole incident resolved because the contract "already exists and runs." What happens: someone, satisfied with this project's clean report, concludes Kiosko already has a complete contract system and needs nothing more. Why it happens: a versioned, parsed contract that generates real tests feels like a finished piece of the system. How to spot it: review this project's "What to expect" final comparison — it's still 3 of 6 dimensions, exactly the same ones module 2 already closed. How to fix it: this project closes the formalization of already-known rules, it adds no new dimension. ORD-9508 (consistency, already resolved in module 3, but outside the scope of orders_contract.yaml as written) and ORD-9509 (accuracy) still need their own tools.

Thinking this module's contract already includes module 3's referential integrity check. What happens: someone expects contract_to_pandera_schema() to also catch ORD-9508 (product_id="P099"), because module 3 already built that check. Why it happens: both modules work on S04's same file, and it's easy to expect the tools to combine automatically. How to spot it: review ColumnContract in lesson 4 — it has no field at all for declaring a relationship with another table, like dim_product. How to fix it: this guide's contract, as designed, describes a single table at a time — the same scope OrdersSchema already had in module 2 —; extending it to declare relationships between tables (some kind of foreign_key: dim_product.product_id inside the YAML) would be a valid, reasonable extension, but it isn't part of this guide. Module 3's validate_referential_integrity() remains a separate piece, which a complete system (module 8's capstone) combines alongside the contract, not inside it.

Forgetting that this project's implicit check_volume isn't yet module 6's real check_volume(). What happens: someone copies this project's row_count_ok = contract.sla.row_count.min <= df.height <= contract.sla.row_count.max line and uses it as if it were the complete, reusable function the rest of the guide needs. Why it happens: the line works, and it produces the correct result for this specific case. How to spot it: it isn't a function — it's a one-off calculation, written directly inside main(), with no name of its own and no reusability. How to fix it: treat it for what it is, an illustrative preview inside this project — module 6 builds check_volume(df, min_rows=5, max_rows=20) as a real, reusable function, with its own structured report, not as a loose line inside another script.

Exercises

Exercise 1 — Run the whole project yourself, from scratch. In a new folder, with kiosko.duckdb (with orders_s04 loaded) and lesson 3's orders_contract.yaml, run python3 project_contract.py. Confirm you see exactly 4 physical rows with an error and 3 dimensions.

Exercise 2 — Extend main() so it also reports, in module 3's same style, which order_id have no known problem under this contract. Using failing_indices and df["order_id"], compute and show the list of order_id that don't appear in the failure report.

all_ids = set(df["order_id"].to_list())
flagged_ids = set(readable["order_id"].to_list())
clean_ids = sorted(all_ids - flagged_ids)
print(f"\norder_id with no problem under this contract: {len(clean_ids)} ({clean_ids})")

order_id with no problem under this contract: 8 (['ORD-9501', 'ORD-9504', 'ORD-9505', 'ORD-9506', 'ORD-9508', 'ORD-9509', 'ORD-9510', 'ORD-9511'])

Exercise 3 — Argue whether contracts/orders_contract.yaml should live in the same repository as Kiosko's code, or in a separate one. In 2-3 sentences, based on the idea that a contract is an artifact "consumable by any system" (the boundary this module's lesson 1 drew against dbt's schema.yml), argue for or against keeping Kiosko's contracts in a repository separate from the rest of the pipeline.

Summary and next step: closing this module

With this project you close module 4. You learned precisely what a data contract is — schema, SLA, and violation policy, all three together, with real industry history behind them —; you wrote S04's real file, with the exact reason behind every field; you parsed it with pydantic, with the guarantee that a malformed contract fails loudly and precisely, never silently; and you built contract_to_pandera_schema(), the function that confirms, with executed evidence identical byte for byte, the module's central promise: the contract generates the tests, the tests no longer live separately from the contract. You also faced the versioning discipline any real artifact needs, and the module's most uncomfortable governance question: if the system had required a contract from day one, S04 wouldn't have been able to sell.

And, along the way, you made clear — again, with the same honesty as earlier modules — the exact limit of what this module solves: three of the six data quality dimensions are covered, and neither ORD-9508 (consistency, which needs another table) nor ORD-9509 (accuracy, which needs a historical baseline) has any way to express itself inside ColumnContract, as designed.

Where you go next. Module 5 — Accuracy and deterministic anomaly detection — finally takes on ORD-9509: the row three consecutive modules — one, two, and now four — let through with no flag at all. You're going to build a price baseline calculated over Kiosko's clean canonical week, and a threshold check, with no Machine Learning at all, capable of catching the dollars-to-cents bug this guide's market warning cited from the very beginning.

Resources

• Pandera — complete official documentation (DataFrameSchema, Field, Check, lazy, SchemaErrors reference). pandera.readthedocs.io. In English.
• Pydantic — complete official documentation (BaseModel, model_validate, aliases, ValidationError). docs.pydantic.dev/latest. In English.
• Module 2, project (lesson 8), of this same guide — the exact comparison baseline for this entire project. src/guides/data-reliability-and-governance-guide/workbook/module-02-declarative-data-quality-tests-with-pandera/en/08-project-testing-s04-with-pandera.md. In English.
• Module 3, project (lesson 8), of this same guide — this module's narrative starting point, including the explicit mention of contracts as the next step. src/guides/data-reliability-and-governance-guide/workbook/module-03-consistency-and-referential-checks/en/08-project-s04s-full-consistency-report.md. In English.
• This guide's DESIGN — the complete map of all eight modules, including the module 5 that follows. src/guides/data-reliability-and-governance-guide/DISENO.md. In Spanish.