PyCharter

PyCharter enforces data contracts where your code runs.

Define a contract once — from a Pydantic model (pycharter contract from-model) or versioned YAML — and the same artifact validates API payloads in-process, gates Kafka consumers record by record, and checks ETL batches at stage boundaries. Deterministic, Pydantic-native, in your process: no warehouse round-trip, no separate CI engine deciding after the fact. The contract is what actually runs.

Most contract tooling stops earlier in the lifecycle: datacontract-cli — the ODCS reference implementation — verifies contracts in CI and batch; Soda and Great Expectations scan the warehouse; dbt tests run after materialisation. PyCharter is the runtime half of that ecosystem: it executes contracts inside the producing service, and round-trips ODCS (v3.1.0 export, v3.0.x / v3.1.x import) so the contracts you enforce at runtime slot into the tools you already use.

Full documentation: https://optophi.github.io/pycharter/

Where PyCharter fits vs datacontract-cli, Pydantic, Pandera, GX, Soda, dbt →

Validate a record in 30 seconds

pip install pycharter

from pycharter import Validator

validator = Validator.from_dict(
    {
        "type": "object",
        "version": "1.0.0",
        "properties": {
            "name": {"type": "string"},
            "age": {"type": "integer"},
        },
        "required": ["name", "age"],
    }
)

result = validator.validate({"name": "Alice", "age": 30})
print(result.is_valid)  # True

result = validator.validate({"name": "Bob", "age": "N/A"})
print(result.errors[0])  # Column 'age': expected integer, found string 'N/A'

That's the library at its smallest — no database, no setup. In production you'd load the same contract from a file (Validator.from_file("contract.yaml")) or a shared store. Failures come back as structured diagnostics (row, column, expected type, offending value), and the generated model validates as fast as a hand-written Pydantic model (~0.7 µs/record — measured).

Prefer the shell? The same check works on a whole file:

pycharter validate contract.yaml data.jsonl   # colorized report, exit 0/1

Why PyCharter

Teams often maintain a Pydantic model at the API, a Pandera schema in the pipeline, a Great Expectations suite for quality, and an ODCS YAML as the "contract" — and they drift. PyCharter unifies authoring and enforcement so the contract is what actually runs, where it runs being the point:

_{● full · ◐ partial · ○ none — markers reflect scope, not quality; see the full comparison}

• One definition, every surface — model or YAML → API gate, stream gate, ETL steps, quality jobs, ODCS round-trip.
• Enforces, not just authors — rejects or coerces bad records before they land; warehouse-side engines measure what did land.
• Deterministic and auditable — versioned contracts, reproducible runs, thread-safe shared validators.

Choose Pydantic alone for one-service API shape checks, Pandera for DataFrame validation in one process, GX/Soda for warehouse-centric batch quality, datacontract-cli for ODCS authoring and CI gates — and PyCharter when the contract must be the enforced single source of truth across services. Full narrative and per-tool sections: PyCharter vs. the alternatives.

Gate a FastAPI endpoint

pip install pycharter[server]

from fastapi import Depends, FastAPI
from pycharter.contrib.fastapi import contract_dependency

app = FastAPI()
validate_order = contract_dependency("contracts/orders.yaml")  # built once


@app.post("/orders")
async def create_order(order: dict = Depends(validate_order)) -> dict:
    # `order` is the coerced + validated payload.
    return {"status": "accepted", "order_id": order["order_id"]}

Contract violations return a structured 422; the same YAML also drives your pipelines and workers. Recipe: Gate a FastAPI endpoint.

Validate a Kafka topic, per record

import asyncio
from pycharter import kafka


async def main():
    async for record in kafka.consume(
        topic="orders",
        bootstrap_servers="localhost:9092",
        group_id="my-service",
        contract_dir="contracts/orders",  # or contract_store=...
    ):
        process(record.payload)  # only records that passed the contract


asyncio.run(main())

Invalid records route to a DLQ with a structured envelope; detect / shadow / enforce modes support safe rollouts, and pycharter stream run runs it as a worker with Prometheus metrics. Start at Streaming — start here.

Gate contract evolution in CI

pycharter contract diff old.yaml new.yaml --fail-on-breaking

Classifies every field/type/required change as breaking or safe (backward-compat semantics) — drop into pre-commit or CI so a contract change can't ship without review.

Concepts

  Option A: no database   schema/contract in code or YAML → Validator → validate(data)
  Option B: with store    contract in DB → Validator(store=...)       → validate(data)
  Option C: pipelines     extract → [contract] → transform → [contract] → load

Start with Option A; add the store when contracts need versioned sharing; add pipeline/stream enforcement where data moves. Deep dives: Concepts · Contract → ETL → quality journey · Cookbook.

Installation

pip install pycharter

That's all you need for in-process validation. Optional extras add capabilities — install only what you use:

Database setup (only for the contract store, API, or UI)

pycharter db init     # creates schema (default: sqlite:///pycharter.db)
pycharter db seed     # loads reference data

Skip this entirely if you only validate dicts or YAML contracts directly. PostgreSQL/SQLite have managed Alembic migrations; see Store backend tiers and the Configuration Guide.

ODCS and the contract ecosystem

PyCharter round-trips the Open Data Contract Standard: to_odcs() emits v3.1.0 documents that validate against the published schema; from_odcs() imports v3.0.x / v3.1.x documents authored anywhere — so contracts written with datacontract-cli enforce at runtime here, and contracts authored here flow back into the ODCS toolchain. PyCharter is not a rival standard; it is an execution engine for the standard.

from pycharter.contract_io import from_odcs, to_odcs
from pycharter import Validator

contract = from_odcs(odcs_document)  # any v3.0.x / v3.1.x doc
validator = Validator(contract)  # enforce it immediately

How PyCharter composes with datacontract-cli, Soda, GX, and dbt: ODCS integration and interop.

Agents and MCP

PyCharter ships an MCP server (python -m pycharter.mcp, stdio JSON-RPC) that serves governed context packs — concepts, relationships, and the contract fields bound to them — to Claude Code, Claude Desktop, or any MCP client, with role/workspace policy filtering. The context an agent receives is bound to the same contracts PyCharter enforces at runtime, so what the agent is told about your data is what your validators actually check.

pip install pycharter[ai]
export PYCHARTER_DATABASE_URL=postgresql://localhost/pycharter
python -m pycharter.mcp

Tools, client config, and the governance model: Agent integration (MCP).

Architecture at a glance

Each service is independent yet composable. The full tour — including the REST API and Web UI wrappers — lives in the Core services reference; the layered design philosophy is in ARCHITECTURE.md.

Picking the right API

The canonical path is the Validator class — create one from a contract file, dict, directory, or store, and call validate(record) / validate_batch(records). It compiles the Pydantic model once, is thread-safe to share, and is what production code should use.

ETL flows use Pipeline; quality checks use QualityCheck; contract storage is pluggable via ContractStoreClient implementations. Built-in coercions/validations and custom registration: reference.

Documentation

• Docs site: https://optophi.github.io/pycharter/ — or pycharter docs serve locally (pip install pycharter[docs])
• Start here · Quick Start tour · Cookbook
• End-to-end user guide · Admin / governance guide
• Streaming — start here · Configuration · Validator benchmarks

Development and testing

python -m venv venv && source venv/bin/activate
pip install -e ".[dev]" && pre-commit install
pytest tests/unit            # fast suite; `pytest -m integration` needs Docker
./scripts/ci.sh              # full GHA-parity gate before opening a PR

See CONTRIBUTING.md for the workflow and AGENTS.md for AI-assisted contributions.

Contributing

Contributions are welcome — see CONTRIBUTING.md. Report security issues per SECURITY.md; community expectations are in CODE_OF_CONDUCT.md.

License

MIT — see LICENSE.

Links

• Repository: github.com/optophi/pycharter
• Issues: GitHub Issues
• PyPI: pypi.org/project/pycharter