PyCharter

Data Contract Management, ETL Pipelines, and Quality Assurance for Python

---

PyCharter is a comprehensive data contract management platform for Python that enables you to define, store, version, enforce, and monitor data contracts throughout your data pipelines.

Three domains

PyCharter is organized around three domains, each with its own storage layer and API surface:

Domain	Purpose	Storage	Key entity
Pipelines	ETL: extract, transform, load. Config-driven or programmatic DAGs.	Pipeline Store	Pipeline config (name + version)
Contracts	Data contracts: schema, coercion/validation rules, metadata. Single source of truth for validation and quality.	Contract Store	Contract (name + version)
Ontology	Semantic layer: concept schemes, concepts, relationships, field bindings. Governance and meaning of data.	Semantic Store	Concept scheme, concept, ontology config

Terminology:

• Contract — A versioned data contract (schema + rules + metadata). Keyed by contract name and contract version.
• Pipeline — A versioned ETL pipeline (steps + settings). Keyed by pipeline name and pipeline version.
• Concept scheme — A named vocabulary of business concepts (e.g. trading-domain). Contains concepts and optional relationships.
• Contract Store — Schema registry: stores contracts and their artifacts (schema, rules, metadata, field mapping). Used by validation and quality.
• Pipeline Store — Pipeline registry: stores pipeline configurations so the API/UI can run and manage them by name/version.
• Semantic Store — Ontology registry: stores concept schemes, concepts, relationships, and versioned ontology configs (e.g. LinkML YAML).

See Concepts and the API Reference — Storage for details.

Key Features

• ETL Pipelines

  ---

  Build data pipelines with a fluent | operator or YAML configs. Built-in extractors for HTTP, files, databases, cloud storage, streaming, and messaging. Use the UI Pipelines section for the ETL generator, run history, and the Pipeline diagram visual editor.

  ETL Tutorial

• Data Contracts

  ---

  Define formal agreements specifying data structure, quality rules, and governance policies. Use a contract-first workflow: define → parse/build → store → validate.

  Contract-First Guide · Contracts Tutorial

• Quality Assurance

  ---

  Monitor data quality with metrics, track violations, and set threshold alerts.

  Quality Tutorial

• Schema Registry

  ---

  Centralized storage for schemas with PostgreSQL, SQLite, MongoDB, or Redis backends.

  Contract Store Tutorial

• Ontology & Semantic Layer

  ---

  Concept schemes, concept types, relationships, field bindings, and a diagrammatic ontology workspace in the UI. Annotate contract fields with business concepts and track semantic health.

  Ontology Guide

• Real-time collaboration

  ---

  Optional Socket.IO relay (pycharter[collab]) for encrypted multi-user sessions in the UI; mounts at /collab when the API runs with the extra installed.

  Collaboration guide

Quick Example

ETL Pipeline with | Operator

import asyncio
from pycharter import Pipeline, HTTPExtractor, PostgresLoader, Rename, Filter

# Build pipeline with fluent syntax
pipeline = (
    Pipeline(HTTPExtractor(url="https://api.example.com/users"))
    | Rename({"user_name": "name", "user_email": "email"})
    | Filter(lambda r: r.get("active", False))
    | PostgresLoader(connection_string="postgresql://...", table="users")
)

# Run the pipeline
result = asyncio.run(pipeline.run())
print(f"Loaded {result.rows_loaded} rows")

Data Validation

from pycharter import Validator

# Create validator from contract file
validator = Validator.from_file("user_contract.yaml")

# Validate data
result = validator.validate({"name": "Alice", "age": 30, "email": "alice@example.com"})

if result.is_valid:
    print(f"Valid: {result.data}")
else:
    print(f"Errors: {result.errors}")

Quality Check

from pycharter import QualityCheck, QualityCheckOptions, QualityThresholds

# Run quality check with thresholds (by contract name/version or pass contract)
check = QualityCheck(store=store)
report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=records,
    options=QualityCheckOptions(check_thresholds=True, thresholds=QualityThresholds(min_overall_score=95.0)),
)

print(f"Quality Score: {report.quality_score.overall_score}/100")
print(f"Passed: {report.passed}")

Installation

Core (validation only)With API serverWith Web UIFull installation

pip install pycharter

pip install pycharter[api]

pip install pycharter[ui]

pip install pycharter[api,ui,etl]

Architecture Overview

graph TB
    subgraph Input["Data Sources"]
        HTTP[HTTP/API]
        Files[Files]
        DB[(Database)]
        Cloud[Cloud Storage]
        Stream[SSE / WebSocket]
        MQ[Kafka / RabbitMQ / SQS]
    end

    subgraph PyCharter["PyCharter"]
        Extract[Extractors]
        Transform[Transformers]
        Load[Loaders]
        Validate[Validator]
        Quality[Quality Check]
        Store[(Contract Store)]
    end

    subgraph Output["Destinations"]
        PG[(PostgreSQL)]
        File[Files]
        S3[Cloud Storage]
    end

    HTTP --> Extract
    Files --> Extract
    DB --> Extract
    Cloud --> Extract
    Stream --> Extract
    MQ --> Extract

    Extract --> Transform
    Transform --> Validate
    Validate --> Load
    Validate --> Quality

    Store --> Validate
    Quality --> Store

    Load --> PG
    Load --> File
    Load --> S3

Next Steps

• Get Started

  ---

  Install PyCharter and run your first pipeline in minutes.

  Installation

• Learn

  ---

  Follow step-by-step tutorials for each major feature.

  Tutorials

• API Reference

  ---

  Detailed documentation for all classes and functions.

  API Reference

• Contribute

  ---

  Help improve PyCharter by contributing code or documentation.

  Contributing