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.

Artifacts

PyCharter has six artifacts. Every one is a YAML file in the same envelope, so all six are authored, versioned, diffed, and reviewed the same way:

api_version: pycharter.io/v1   # the format marker — always this
kind: DataContract             # which artifact this is (one of the six below)
metadata:
  name: orders                 # identity slug
  namespace: example           # identity scope (optional; absent = global)
  version: "1.0.0"             # content version
spec:                          # the body — shape depends on `kind`
  ...

Identity is (metadata.namespace, metadata.name, metadata.version), and name is also the display label — there is no separate title field.

Scaffold one with pycharter contract init, pycharter pipeline init, pycharter stream init, or pycharter bridge init; runnable end-to-end examples live in data/seed/examples/. Files written by an older PyCharter still load, and pycharter migrate rewrites them in place — see Artifact format versioning.

1. DataContract

The artifact every enforcement point loads. One file carries the shape (json_schema), the transforms applied before validation (coercion_rules), and the business rules applied after the types check out (validation_rules):

# contracts/orders.yaml
api_version: pycharter.io/v1
kind: DataContract
metadata:
  name: order
  version: "1.0.0"
  status: active
  type: domain_entity
  description: A customer order accepted by the storefront API.
  ownership:
    roles:
      business_owner: [storefront-team]
      technical_owner: [data-platform]

spec:
  json_schema:
    type: object
    title: order
    version: "1.0.0"
    additionalProperties: false
    required: [order_id, customer_id, total_amount, currency_code, status]
    properties:
      order_id:      {type: string, format: uuid}
      customer_id:   {type: string}
      total_amount:  {type: number, minimum: 0}
      currency_code: {type: string, minLength: 3, maxLength: 3}
      status:        {type: string}
      placed_at:     {type: string, format: date-time}

  # Before validation: make messy input typed. "19.99" -> 19.99, "usd" -> "USD".
  coercion_rules:
    version: "1.0.0"
    rules:
      total_amount: coerce_to_float
      currency_code: coerce_to_uppercase
      placed_at: coerce_to_datetime

  # After validation: business rules beyond types.
  validation_rules:
    version: "1.0.0"
    rules:
      total_amount:
        greater_than_or_equal_to: {threshold: 0}
      currency_code:
        matches_regex: {pattern: "^[A-Z]{3}$"}
      status:
        only_allow:
          allowed_values: [pending, paid, shipped, cancelled]

That is the contracts/orders.yaml the FastAPI gate above loads — and the same file backs pycharter validate contracts/orders.yaml orders.jsonl, the Kafka consumer, and the pipeline below. Rules are named, not code: 21 built-in coercions and 16 validations, listed in built-in coercions and validations (add severity: warning inside a rule to report instead of reject). Start one from a real sample with pycharter contract sketch --from-json sample.json -o contracts/orders.yaml.

A contract's four body sections — json_schema, coercion_rules, validation_rules, and field_bindings (below) — can also be stored separately and reused by reference, so two contracts share one set of validation rules instead of copying them. Authoring them inline, as above, is the default and always works.

2. Pipeline

Batch extract → transform → load. The contract: block attached to a step is the enforcement point — put it on whichever steps must be gated:

# pipelines/orders_ingest/pipeline.yaml
api_version: pycharter.io/v1
kind: Pipeline
metadata:
  name: orders_ingest
  namespace: examples
  version: "1.0.0"
  description: Load the daily orders CSV, normalize it, export validated JSONL.

spec:
  variables:
    INPUT_PATH: data/input/orders.csv
    OUTPUT_PATH: data/exports/orders.jsonl

  steps:
    - id: extract_orders
      type: extract
      source:
        type: file
        path: ${INPUT_PATH}
        format: csv

    - id: normalize_orders
      type: transform
      input: extract_orders
      operations:
        - rename: {order_total: total_amount}
        - defaults: {status: pending}

    - id: load_orders
      type: load
      input: normalize_orders
      target:
        type: file
        path: ${OUTPUT_PATH}
        format: jsonl
        mode: overwrite
      contract:                      # <- the enforcement point
        ref: {type: store, name: order, version: "1.0.0"}
        coerce: true                 # apply coercion_rules
        validate: true               # reject records that fail the contract
        quality: true                # run the contract's quality checks

pycharter pipeline run ./pipelines/orders_ingest --watch

Use ref: {type: file, path: ../../contracts/orders.yaml} to point at a contract on disk instead of the store. Attach the block to any step — gate on extract, after transform, before load, or all three. Sources include file, http, and SQL; ${VAR} placeholders resolve from the spec.variables: block first, then the environment, with ${VAR:-default} and ${VAR:?required} semantics. More: Building pipelines.

3. StreamingWorker

The same contract, enforced record by record on a live stream instead of a batch. mode is what makes a rollout safe: detect passes every record through and only counts violations, shadow forwards invalid records and copies them to the DLQ, and enforce (the default) sends them to the DLQ only:

# streams/orders/stream-worker.yaml
api_version: pycharter.io/v1
kind: StreamingWorker
metadata:
  name: orders
  namespace: example
  version: "1.0.0"
  description: Validate the orders.raw topic against the order contract.

spec:
  variables:
    KAFKA_BROKERS: localhost:9092

  streams:
    - name: orders-validator
      source:
        type: kafka                  # kafka | websocket | sse | rabbitmq | sqs
        topic: orders.raw
        bootstrap_servers: ${KAFKA_BROKERS}
        consumer_group: pycharter-orders
        auto_offset_reset: latest
      contract:                      # same block shape as a pipeline step
        ref: {type: store, name: order, version: "1.0.0"}
        coerce: true
        validate: true
      mode: enforce                  # detect | shadow | enforce
      delivery: at_least_once
      load:
        target:
          type: kafka
          topic: orders.validated
          bootstrap_servers: ${KAFKA_BROKERS}
      dlq:
        target:
          type: kafka
          topic: orders.dlq
          bootstrap_servers: ${KAFKA_BROKERS}

  observability:
    health_port: 8080                # /healthz, /readyz, /metrics
    log_format: json

pycharter stream run ./streams/orders/stream-worker.yaml

More: Streaming — start here.

4. Bridge

A WebSocket is a single connection, so running more workers against one just duplicates consumption. A bridge republishes the firehose onto a partitioned Kafka topic, which a scaled StreamingWorker group can then consume. It only moves bytes — no contract, no validation, no DLQ:

# bridges/firehose/bridge.yaml
api_version: pycharter.io/v1
kind: Bridge
metadata:
  name: firehose-bridge
  namespace: example
  version: "1.0.0"
  description: Republish a WebSocket firehose to a Kafka topic.

spec:
  variables:
    KAFKA_BROKERS: localhost:9092

  bridges:
    - name: firehose
      source:
        type: websocket
        url: ${WS_URL:?WS_URL is required}
        data_format: json
        ping_interval: 20.0          # keepalive seconds
        idle_timeout: 60.0           # reconnect after this much silence
      sink:
        type: kafka
        topic: market.raw
        bootstrap_servers: ${KAFKA_BROKERS}

pycharter bridge run ./bridges/firehose/bridge.yaml

5. ConceptScheme

Contracts describe the shape of a field. A concept scheme describes what fields mean, once, for every contract that uses them:

# vocabulary/ecommerce.yaml
api_version: pycharter.io/v1
kind: ConceptScheme
metadata:
  name: ecommerce
  version: "2.0.0"
  description: Governed vocabulary for the storefront domain.

spec:
  concepts:
    - id: Customer
      label: Customer
      concept_type: entity          # a thing with identity
      definition: A person or organization that places orders.
    - id: Order
      label: Order
      concept_type: event
      definition: A customer's request to purchase one or more products.
    - id: MonetaryAmount
      label: Monetary Amount
      concept_type: attribute       # a property or measurement
      definition: A value of money in a stated currency.
    - id: OrderTotal
      label: Order Total
      concept_type: attribute
      broader: MonetaryAmount       # narrower than MonetaryAmount
      definition: Amount payable for an order, including tax and shipping.

Contracts then bind their fields to those concepts, in the same contract file as the schema:

spec:
  field_bindings:
    order_id:
      primary:
        concept: {concept: Order}
        concept_scheme: "ecommerce:2.0.0"
        role: identity              # this field identifies the record's entity
    customer_id:
      primary:
        concept: {concept: Customer}
        concept_scheme: "ecommerce:2.0.0"
        role: reference             # points at a different entity
    total_amount:
      primary:
        concept: {concept: OrderTotal}
        concept_scheme: "ecommerce:2.0.0"
        role: attribute             # a property of this record (the default)
      annotations:                  # optional extra meaning, zero or more
        - concept: MonetaryAmount
          concept_scheme: "ecommerce:2.0.0"
          relationship: derived_from

A binding may also carry pii: true (personal data) or restricted: true (raw values must never be captured) — field-level flags that govern what downstream profiling, extraction, and AI surfaces are allowed to see.

Bindings are what make two contracts that both call a column total comparable, and they are what the MCP server serves to agents: the context an agent receives about a field is bound to the same contract your validators enforce. A field may stay unbound — governance tooling simply flags it for review.

6. ConceptSchemeTemplate

A starter vocabulary — the same idea as a ConceptScheme, but shipped as a reusable pattern you copy and adapt rather than a governed scheme you enforce against. PyCharter bundles six (ecommerce, healthcare, saas, party, product_catalog, event_participation); this is the shape if you write your own:

# templates/subscriptions.yaml
api_version: pycharter.io/v1
kind: ConceptSchemeTemplate
metadata:
  name: subscriptions
  version: "1.0.0"
  description: Tenants, plans, and recurring charges.
  category: domain               # domain (a full domain) | pattern (a reusable cluster)
  tags: [billing, saas]

spec:
  concepts:
    - name: Tenant
      concept_type: entity
      definition: An organization that holds a subscription.
      alt_labels: [Account, Workspace]
    - name: Plan
      concept_type: reference_data
      definition: A named tier a tenant can subscribe to.
    - name: Subscription
      concept_type: event
      definition: A tenant's commitment to a plan over a billing period.

  relationships:
    - {source: Tenant, target: Subscription, type: has, label: holds}
    - {source: Subscription, target: Plan, type: references, label: on plan}

Browse and fetch the bundled ones with GET /api/v1/semantic/templates, or read them under data/templates/concept_schemes/.

More on the vocabulary layer: Vocabulary · Binding any resource to a concept · a complete worked example (22 contracts, a scheme, and a pipeline) ships in data/seed/examples/ecommerce/.

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