pycharter 0.0.36
A Python package for data contract management with five core services: contract parsing, metadata storage, Pydantic generation, JSON Schema conversion, and runtime validation
Requires Python: >=3.10
Keywords: semantic, python, pydantic, contract, validation, schema

# PyCharter

> **Data contract management and validation for Python: define schemas, enforce quality, and run ETL with contracts.**

[![Python 3.10+](https://img.shields.io/badge/python-3.10+-blue.svg)](https://www.python.org/downloads/)
[![License: MIT](https://img.shields.io/badge/License-MIT-yellow.svg)](https://opensource.org/licenses/MIT)
[![Code style: black](https://img.shields.io/badge/code%20style-black-000000.svg)](https://github.com/psf/black)

---

## ⚡ Quick start (2 minutes)

Install, define a tiny contract, and validate one record. Copy-paste into a new terminal:

```bash
pip install pycharter
```

```python
from pycharter import from_dict, validate

# 1. Define a minimal contract (schema)
schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
    },
    "required": ["name", "age"],
}

# 2. Build a validator and validate one record
Person = from_dict(schema, "Person")
result = validate(Person, {"name": "Alice", "age": 30})

print(result.is_valid)   # True
print(result.data.name) # Alice
```

Invalid data returns `result.is_valid == False` and `result.errors` with details. Next: use a **contract file** with `Validator.from_file("contract.yaml")`, or add **coercion/validation rules** and store contracts in a **metadata store** (see [Concepts](#-concepts) and [Installation](#-installation)).

---

## What is PyCharter?

**PyCharter** is a **Data Contract Management and Quality Assurance** platform for Python that enables you to define, store, version, enforce, and monitor data contracts throughout your data pipelines. At its core, PyCharter automatically converts JSON schemas into fully-functional Pydantic models, fully supporting the JSON Schema Draft 2020-12 standard, including all standard validation keywords (minLength, maxLength, pattern, enum, minimum, maximum, etc.), while also providing extensions for pre-validation coercion and post-validation checks. It handles nested objects, arrays, and custom validators, with all validation logic stored as data (not Python code). PyCharter also includes a comprehensive quality assurance module for monitoring data quality, tracking violations, and generating quality reports.

## What is PyCharter?

PyCharter is a **data contract-as-code** platform that brings structure, validation, and governance to data pipelines and applications. It provides:

- **Data Contract Definition & Management**: Define formal agreements that specify data structure, quality rules, and governance policies
- **Schema Registry**: Centralized storage and versioning of data schemas with support for schema evolution
- **Data Quality Enforcement**: Coercion rules (data transformation) and validation rules (business constraints) to ensure data integrity
- **Data Governance**: Track ownership, stewardship, and enforce data policies across your organization
- **Metadata Management**: Store and retrieve data about data (schemas, ownership, rules, lineage)
- **Runtime Validation**: Validate data against contracts in production pipelines, APIs, and ETL processes

### Data Terminology

PyCharter implements key data management concepts:

- **Data Contract**: Formal specification of data structure, quality rules, and governance policies that define the "contract" between data producers and consumers
- **Schema Registry**: Centralized repository for storing and versioning data schemas, enabling schema discovery and evolution tracking
- **Data Quality**: Coercion (pre-validation transformation) and validation (post-validation checks) rules that ensure data integrity
- **Data Governance**: Ownership tracking, stewardship management, and policy enforcement for data assets
- **Schema Evolution**: Versioning and migration capabilities that allow data structures to evolve over time while maintaining backward compatibility
- **Metadata Management**: Storage and retrieval of data about data, including schemas, ownership information, governance rules, and lineage

### Use Cases

- **Data Pipeline Validation**: Ensure data conforms to contracts before processing in ETL pipelines
- **API Contract Enforcement**: Validate API request/response data against defined contracts
- **Data Integration**: Standardize data formats across systems and services
- **Compliance & Governance**: Track ownership, enforce data policies, and maintain audit trails
- **Schema Registry**: Centralized schema management for microservices and data platforms
- **Data Quality Assurance**: Catch data quality issues early in the pipeline through automated validation

---

## 📐 Concepts

A short mental model so you know what to reach for.

| Concept | What it is | When you use it |
|--------|------------|------------------|
| **Schema** | The shape of the data (JSON Schema): types, required fields, nested objects. | When you only need structure (e.g. “this field is string, that one is integer”). |
| **Data contract** | Schema + **coercion rules** (e.g. string → int) + **validation rules** (e.g. min/max, allowed values) + optional **metadata** (ownership, governance). | When you want one artifact that defines structure, transforms, and business rules. |
| **Metadata store** | A database (SQLite, PostgreSQL, etc.) that stores contracts (and their versions) so many apps can reuse them. | When you have multiple services or pipelines and want a single source of truth. |
| **ETL validation** | Validating data **after extract** (source) and **before load** (target) in a pipeline, using a schema or contract. | When you run ETL and want to reject or quarantine bad rows at stage boundaries. |

Flow from “just validate” to “contracts in a store and ETL”:

```
  ┌─────────────────────────────────────────────────────────────────────────┐
  │  Option A: No database                                                   │
  │  Schema/contract in code or YAML file  →  Validator  →  validate(data)  │
  └─────────────────────────────────────────────────────────────────────────┘

  ┌─────────────────────────────────────────────────────────────────────────┐
  │  Option B: With metadata store                                           │
  │  Contract in DB  →  Validator(store=..., schema_id=...)  →  validate()  │
  └─────────────────────────────────────────────────────────────────────────┘

  ┌─────────────────────────────────────────────────────────────────────────┐
  │  Option C: ETL pipeline                                                  │
  │  Extract  →  [optional: validate with contract]  →  Transform  →         │
  │  [optional: validate with contract]  →  Load                            │
  └─────────────────────────────────────────────────────────────────────────┘
```

Start with **Option A** (Quick start above); add **metadata store** when you need versioned, shared contracts; add **ETL validation** when you run pipelines and want contract checks at extract/load.

---

## ✨ Features

- 🚀 **Dynamic Model Generation** - Convert JSON schemas to Pydantic models at runtime
- 📋 **JSON Schema Compliant** - Full support for JSON Schema Draft 2020-12 standard
- 🔄 **Type Coercion** - Automatic type conversion before validation (e.g., string → integer)
- ✅ **Custom Validators** - Built-in and extensible validation rules
- 🏗️ **Nested Structures** - Full support for nested objects and arrays
- 📦 **Multiple Input Formats** - Load schemas from dicts, JSON strings, files, or URLs
- 🎯 **Type Safe** - Full type hints and Pydantic v2 compatibility
- 🔧 **Extensible** - Register custom coercion and validation functions
- 📊 **Data-Driven** - All validation logic stored as JSON data, not Python code
- 📝 **Data Contract Management** - Complete lifecycle management for data contracts with versioning
- 💾 **Schema Registry** - Centralized schema storage with support for PostgreSQL, MongoDB, Redis, and InMemory
- 🏛️ **Data Governance** - Track ownership, stewardship, and enforce governance policies
- 🔍 **Metadata Management** - Store and query metadata about your data assets
- 📈 **Schema Evolution** - Version schemas and track changes over time
- ✅ **Quality Assurance** - Monitor data quality, calculate metrics, track violations, and generate reports
- 📊 **Quality Metrics** - Calculate quality scores, accuracy, completeness, and violation rates
- 🚨 **Threshold Alerting** - Set quality thresholds and get alerts when quality degrades

## 📦 Installation

### Core Library

```bash
pip install pycharter
```

### With API Support

```bash
pip install pycharter[api]
```

This installs FastAPI and Uvicorn for running the REST API server.

### With UI Support

```bash
pip install pycharter[ui]
```

This installs the Python dependencies and **pre-built UI static files** (like Airflow).

**After installation, you can immediately start the UI:**

```bash
pycharter ui serve    # Production mode (uses pre-built static files)
```

**For development** (if you have the source code):

```bash
cd ui
npm install          # Install Node.js dependencies
pycharter ui dev     # Development mode with hot reload
```

**Note**: When installed from pip, the UI works immediately without Node.js. For development, Node.js is required. See `ui/INSTALLATION.md` for detailed instructions.

### Database setup (for metadata store, API, and seed data)

If you use the **metadata store** (SQLite or PostgreSQL), the **REST API**, or **seed data**, initialize the database once:

```bash
# Initialize database schema (SQLite default: sqlite:///pycharter.db)
pycharter db init

# Seed reference data (owners, domains, systems, environments, data feeds, compliance frameworks, tags)
pycharter db seed
```

**From the project source** (e.g. after cloning the repo), you can run the full seed script to also load example aviation contracts into the metadata store:

```bash
python bin/seed                    # seed data/seed + data/aviation_examples
python bin/seed --skip-aviation    # seed only data/seed
```

Use `pycharter db init --force` to drop and recreate all tables (SQLite only; **destroys existing data**). See [Configuration Guide](docs/guides/configuration.md) for connection options and migrations.

## 🚀 Quick Start

### Quick Start: ETL Pipelines

Build and run ETL pipelines programmatically (with the `|` operator) or from YAML configs. Pipeline **run()** is async; use `asyncio.run()` from scripts or `await` in async code.

```python
import asyncio
from pycharter import Pipeline, HTTPExtractor, PostgresLoader, Rename, AddField

# Programmatic pipeline
pipeline = (
    Pipeline(HTTPExtractor(url="https://api.example.com/data"))
    | Rename({"old": "new"})
    | AddField("processed_at", "now()")
    | PostgresLoader(connection_string="...", table="users")
)
result = asyncio.run(pipeline.run())

# Config-driven: explicit files
pipeline = Pipeline.from_config_files(
    extract="configs/extract.yaml",
    load="configs/load.yaml",
    variables={"API_KEY": "secret"}
)

# Config-driven: directory (extract.yaml, transform.yaml, load.yaml)
pipeline = Pipeline.from_config_dir("pipelines/users/")

# Config-driven: single file
pipeline = Pipeline.from_config_file("pipelines/users/pipeline.yaml")

result = asyncio.run(pipeline.run())
```

See **ETL Pipelines** under Core Services for error handling (`error_context`, `ErrorMode`) and variable resolution (`PipelineContext(variables={...})`).

### Quick Start: Convenience Functions (One-off Use)

```python
from pycharter import from_dict, validate

# Define your JSON schema
schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "name": {"type": "string"},
        "age": {"type": "integer"},
        "email": {"type": "string"}
    },
    "required": ["name", "age"]
}

# Generate a Pydantic model (convenience function)
Person = from_dict(schema, "Person")

# Validate data
result = validate(Person, {"name": "Alice", "age": 30, "email": "alice@example.com"})
if result.is_valid:
    print(f"Valid: {result.data.name}")  # Output: Valid: Alice
```

### Production Use: Validator Class (Recommended)

For production code with multiple validations, use the `Validator` class for better performance. Create validators via **factory methods** or from a metadata store:

```python
from pycharter import Validator

# From directory (expects schema.yaml, coercion_rules.yaml, validation_rules.yaml)
validator = Validator.from_dir("data/contracts/user")

# From explicit files (any filenames)
validator = Validator.from_files(
    schema="schemas/user.yaml",
    coercion_rules="rules/coercion.yaml",
    validation_rules="rules/validation.yaml"
)

# From a single contract file
validator = Validator.from_file("user_contract.yaml")

# From dictionaries
validator = Validator.from_dict(schema={...}, coercion_rules={...}, validation_rules={...})

# From metadata store (with database)
validator = Validator(store=store, schema_id="user_schema_v1")

# Validate multiple records efficiently (model is cached)
result1 = validator.validate({"name": "Alice", "age": 30})
result2 = validator.validate({"name": "Bob", "age": 25})

# Batch validation
results = validator.validate_batch([data1, data2, data3])
```

### With Metadata Store

```python
from pycharter import Validator, SQLiteMetadataStore

# Connect to metadata store
store = SQLiteMetadataStore("metadata.db")
store.connect()

# Create validator from store
validator = Validator(store=store, schema_id="user_schema_v1")

# Validate data
result = validator.validate({"name": "Alice", "age": 30})
```

## 📐 API Organization

PyCharter's API is organized into **three tiers** to help you choose the right approach for your use case:

### Tier 1: Primary Interfaces (⭐ Recommended for Production)

**Classes** that provide the best performance and most features:
- **`Validator`** - Primary validation interface (use for multiple validations); create via `from_dir()`, `from_files()`, `from_file()`, `from_dict()` or from store
- **`Pipeline`** - ETL pipeline (programmatic or config-driven); create via `from_config_files()`, `from_config_dir()`, `from_config_file()` or constructor
- **`QualityCheck`** - Primary quality assurance interface
- **`MetadataStoreClient`** - Base class for metadata stores

**When to use**: Production code, batch processing, when you need to validate multiple records or run ETL pipelines.

### Tier 2: Convenience Functions (Quick Start)

**Functions** that make common tasks easy and discoverable:
- **Input helpers**: `from_dict()`, `from_file()`, `from_json()`, `from_url()`
- **Output helpers**: `to_dict()`, `to_file()`, `to_json()`
- **Validation helpers**: `validate_with_store()`, `validate_with_contract()`
- **Contract helpers**: `parse_contract_file()`, `build_contract()`

**When to use**: Quick scripts, one-off validations, exploratory work, learning the library.

### Tier 3: Low-Level Utilities

**Functions** for when you already have models or need fine-grained control:
- **`validate()`** - Validate with existing Pydantic model
- **`validate_batch()`** - Batch validate with existing model
- **`model_to_schema()`** - Core conversion function

**When to use**: Advanced use cases, when you've already generated models, custom workflows.

### Choosing the Right Approach

| Use Case | Recommended Approach | Example |
|----------|---------------------|---------|
| **Production pipeline with multiple validations** | `Validator` class | `validator = Validator(store=store, schema_id="schema"); validator.validate(data)` |
| **Quick one-off validation** | Convenience function | `validate_with_contract("contract.yaml", data)` |
| **You already have a model** | Low-level function | `validate(UserModel, data)` |
| **Batch processing** | `Validator.validate_batch()` | `validator.validate_batch([data1, data2, data3])` |

## 🏗️ Core Services & Data Production Journey

PyCharter provides **eight core services** that work together to support a complete data production journey, from contract specification to quality assurance. Each service plays a critical role in managing data contracts and ensuring data quality throughout your pipeline.

### The Data Production Journey

The typical data production workflow follows this path:

```
1. Data Contract Specification
   ↓
2. Contract Parsing
   ↓
3. Metadata Storage
   ↓
4. Pydantic Model Generation
   ↓
5. Runtime Validation
   ↓
6. Quality Assurance & Monitoring
```

### 1. 📄 Contract Parser (`pycharter.contract_parser`)

**Purpose**: Reads and decomposes data contract files into structured metadata components.

**When to Use**: At the beginning of your data production journey, when you have data contract files (YAML or JSON) that need to be processed and understood.

**How It Works**:
- Accepts data contract files containing schema definitions, governance rules, ownership information, and metadata
- Decomposes the contract into distinct components: `schema`, `governance_rules`, `ownership`, and `metadata`
- Returns a `ContractMetadata` object that separates concerns and makes each component accessible
- Extracts and tracks versions of all components

**Example**:
```python
from pycharter import parse_contract_file, ContractMetadata

# Parse a contract file (YAML or JSON)
metadata = parse_contract_file("data_contract.yaml")

# Access decomposed components
schema = metadata.schema              # JSON Schema definition
governance = metadata.governance_rules # Governance policies
ownership = metadata.ownership         # Owner/team information
metadata_info = metadata.metadata      # Additional metadata
versions = metadata.versions          # Component versions
```

**Contribution to Journey**: The contract parser is the **entry point** that takes raw contract specifications and prepares them for downstream processing. It ensures that contracts are properly structured and that all components (schema, governance, ownership) are separated for independent handling.

---

### 1b. 🏗️ Contract Builder (`pycharter.contract_builder`)

**Purpose**: Constructs consolidated data contracts from separate artifacts (schema, coercion rules, validation rules, metadata).

**When to Use**: When you have separate artifacts stored independently and need to combine them into a single consolidated contract for runtime validation or distribution.

**How It Works**:
- Takes separate artifacts (schema, coercion rules, validation rules, metadata, ownership, governance rules)
- Merges coercion and validation rules into the schema
- Tracks versions of all components
- Produces a consolidated contract suitable for runtime validation
- Can build from artifacts directly or retrieve from metadata store

**Example**:
```python
from pycharter import build_contract, build_contract_from_store, ContractArtifacts

# Build from separate artifacts
artifacts = ContractArtifacts(
    schema={"type": "object", "version": "1.0.0", "properties": {...}},
    coercion_rules={"version": "1.0.0", "rules": {"age": "coerce_to_integer"}},
    validation_rules={"version": "1.0.0", "rules": {"age": {"is_positive": {...}}}},
    metadata={"version": "1.0.0", "description": "User contract"},
    ownership={"owner": "data-team", "team": "engineering"},
)

contract = build_contract(artifacts)
# Contract now has:
# - schema (RAW - rules NOT merged into it)
# - coercion_rules, validation_rules (separate)
# - metadata, ownership, governance_rules
# - versions tracking all components

# Or build from metadata store
contract = build_contract_from_store(store, "user_schema_v1")

# Use for validation - Validator merges rules internally
from pycharter import validate_with_contract
result = validate_with_contract(contract, {"name": "Alice", "age": "30"})
```

**Contribution to Journey**: The contract builder is the **consolidation layer** that combines separate artifacts (stored independently in the database) into a single contract artifact. The contract contains **raw schema + separate rules** (not merged). The Validator class handles merging internally during validation, keeping the contract structure clear and editable.

---

### 2. 💾 Metadata Store Client (`pycharter.metadata_store`)

**Purpose**: Manages persistent storage and retrieval of decomposed metadata in databases.

**When to Use**: After parsing contracts, when you need to store metadata components (schemas, governance rules, ownership) in a database for versioning, querying, and governance.

**How It Works**:
- Provides methods to store and retrieve schemas, governance rules, ownership information, and metadata
- Supports versioning and querying of stored metadata
- Multiple implementations available: PostgreSQL, MongoDB, Redis, and In-Memory (for testing)

**Available Implementations**:
- **PostgresMetadataStore** - For PostgreSQL databases (recommended for production)
- **SQLiteMetadataStore** - For SQLite databases (great for development and small deployments)
- **MongoDBMetadataStore** - For MongoDB databases
- **RedisMetadataStore** - For Redis databases
- **InMemoryMetadataStore** - For testing and development (no persistence)

**Example**:
```python
from pycharter import SQLiteMetadataStore, parse_contract_file

# Parse contract
metadata = parse_contract_file("contract.yaml")

# Use SQLite metadata store (or PostgresMetadataStore, MongoDBMetadataStore, RedisMetadataStore, etc.)
store = SQLiteMetadataStore("metadata.db")
store.connect()

# Store decomposed components
schema_id = store.store_schema("user_schema", metadata.schema, version="1.0")

# Merge ownership and governance into metadata before storing
# Ownership and governance are part of metadata, not separate entities
metadata_dict = metadata.metadata.copy() if metadata.metadata else {}
if metadata.ownership:
    metadata_dict["business_owners"] = [metadata.ownership.get("owner", "unknown")] if metadata.ownership.get("owner") else []
if metadata.governance_rules:
    metadata_dict["governance_rules"] = metadata.governance_rules

# Store metadata once with all information (ownership and governance included)
store.store_metadata(resource_id=schema_id, resource_type="schema", metadata=metadata_dict)

# Store coercion and validation rules
store.store_coercion_rules(schema_id, {"age": "coerce_to_integer"}, version="1.0")
store.store_validation_rules(schema_id, {"age": {"is_positive": {}}}, version="1.0")

# Retrieve later
stored_schema = store.get_schema(schema_id)
coercion_rules = store.get_coercion_rules(schema_id)
validation_rules = store.get_validation_rules(schema_id)
```

**Contribution to Journey**: The metadata store is the **persistence layer** that ensures contracts and their components are versioned, searchable, and accessible across your organization. It enables governance, audit trails, and schema evolution tracking.

**See [Configuration Guide](docs/guides/configuration.md) for database setup and initialization instructions.**

---

### 3. 🏭 Pydantic Generator (`pycharter.pydantic_generator`)

**Purpose**: Dynamically generates fully-functional Pydantic models from JSON Schema definitions.

**When to Use**: After storing schemas (or directly from parsed contracts), when you need to generate Python models for type-safe data validation and processing.

**How It Works**:
- Takes JSON Schema definitions (from contracts or metadata store)
- Programmatically generates Pydantic model classes at runtime
- Supports all JSON Schema Draft 2020-12 features plus custom coercions and validations
- Can generate models from dictionaries, JSON strings, files, or URLs
- Optionally generates Python files with model definitions

**Example**:
```python
from pycharter import from_dict, generate_model_file, MetadataStoreClient

# Option 1: Generate from parsed contract
metadata = parse_contract_file("contract.yaml")
UserModel = from_dict(metadata.schema, "User")

# Option 2: Generate from stored schema
client = MetadataStoreClient(...)
schema = client.get_schema("user_schema_v1")
UserModel = from_dict(schema, "User")

# Option 3: Generate and save to file
generate_model_file(schema, "user_model.py", "User")
```

**Contribution to Journey**: The Pydantic generator is the **transformation engine** that converts declarative JSON Schema definitions into executable Python models. It bridges the gap between contract specifications (data) and runtime validation (code), enabling type-safe data processing.

---

### 4. 🔄 JSON Schema Converter (`pycharter.json_schema_converter`)

**Purpose**: Converts existing Pydantic models back into JSON Schema format (reverse conversion).

**When to Use**: When you have existing Pydantic models and need to generate JSON Schema definitions, or when you want to round-trip between schemas and models.

**How It Works**:
- Takes Pydantic model classes as input
- Generates JSON Schema dictionaries that represent the model structure
- Preserves validation rules, types, and constraints
- Can output to dictionaries, JSON strings, or files

**Example**:
```python
from pycharter import to_dict, to_file, to_json
from pydantic import BaseModel

class Product(BaseModel):
    name: str
    price: float
    in_stock: bool = True

# Convert to JSON Schema
schema = to_dict(Product)
json_string = to_json(Product)
to_file(Product, "product_schema.json")

# Now you can use the schema with other services
ProductModel = from_dict(schema, "Product")  # Round-trip
```

**Contribution to Journey**: The JSON Schema converter enables **bidirectional conversion** between models and schemas. It's useful for:
- Generating schemas from existing code
- Round-trip validation (schema → model → schema)
- Integrating with systems that require JSON Schema format
- Documenting existing models as schemas

---

### 5. ✅ Runtime Validator (`pycharter.runtime_validator`)

**Purpose**: Lightweight validation utility for validating data against generated Pydantic models in production data pipelines.

**When to Use**: In your data processing scripts, ETL pipelines, API endpoints, or any place where you need to validate incoming data against contract specifications.

**API Organization**:

PyCharter provides validation through three tiers:

1. **Tier 1: Validator Class** (⭐ **PRIMARY INTERFACE** - Recommended for production)
   - Best performance for multiple validations (model is cached)
   - Supports all data sources (contract files, directories, stores, dictionaries)
   - Reusable instance for batch processing

2. **Tier 2: Convenience Functions** (Quick start - one-off validations)
   - `validate_with_store()` - Quick validation with metadata store
   - `validate_with_contract()` - Quick validation with contract file/dict
   - `get_model_from_store()` / `get_model_from_contract()` - Get model for reuse

3. **Tier 3: Low-Level Functions** (When you already have a model)
   - `validate()` - Validate single record with existing model
   - `validate_batch()` - Batch validate with existing model

**How It Works**:
- Takes a Pydantic model (generated from a schema) and raw data
- Validates data against the model's constraints
- Returns a `ValidationResult` with validation status, validated data, and errors
- Supports single record and batch validation
- Can be used in strict mode (raises exceptions) or lenient mode (returns results)

**Example - Validator Class (Recommended)**:
```python
from pycharter import Validator, SQLiteMetadataStore

# Option 1: From directory (schema.yaml, coercion_rules.yaml, validation_rules.yaml)
validator = Validator.from_dir("data/contracts/user")
result = validator.validate({"name": "Alice", "age": 30})

# Option 2: From explicit files
validator = Validator.from_files(schema="schemas/user.yaml", coercion_rules="rules/coercion.yaml")
result = validator.validate({"name": "Alice", "age": 30})

# Option 3: From single contract file
validator = Validator.from_file("user_contract.yaml")
result = validator.validate({"name": "Alice", "age": 30})

# Option 4: From metadata store (with database)
store = SQLiteMetadataStore("metadata.db")
store.connect()
validator = Validator(store=store, schema_id="user_schema_v1")
result = validator.validate({"name": "Alice", "age": 30})

# Batch validation (efficient - model cached)
results = validator.validate_batch([data1, data2, data3])
```

**Example - Convenience Functions (Quick Start)**:
```python
from pycharter import validate_with_store, validate_with_contract, SQLiteMetadataStore

# Quick validation with store
store = SQLiteMetadataStore("metadata.db")
store.connect()
result = validate_with_store(store, "user_schema_v1", {"name": "Alice", "age": 30})

# Quick validation with contract file (no database)
result = validate_with_contract("user_contract.yaml", {"name": "Alice", "age": 30})
```

**Example - Low-Level (When You Have a Model)**:
```python
from pycharter import from_dict, validate, validate_batch

# Generate model
UserModel = from_dict(schema, "User")

# Validate single record
result = validate(UserModel, {"name": "Alice", "age": 30})

# Batch validate
results = validate_batch(UserModel, [data1, data2, data3])
```

**Performance Tips**:
- ⚡ **For multiple validations**: Use `Validator` class (model is cached)
- ⚡ **For one-off validations**: Convenience functions are fine
- ⚡ **For batch processing**: Use `Validator.validate_batch()` or `validate_batch()`

**Contribution to Journey**: The runtime validator is the **enforcement layer** that ensures data quality in production. It validates actual data against contract specifications, catching violations early and preventing bad data from propagating through your systems. It supports both database-backed workflows (for production systems with metadata stores) and contract-based workflows (for simpler use cases without database dependencies).

---

### 5b. 🔄 ETL Pipelines (`pycharter.etl_generator`)

**Purpose**: Build and run ETL pipelines programmatically (with the `|` operator) or from YAML configs. No assumptions about project layout—you specify file paths or use a directory with standard filenames.

**When to Use**: When you need to extract, transform, and load data from config-driven or code-defined pipelines (HTTP, files, databases, cloud storage → transforms → Postgres, files, cloud).

**How It Works**:
- **Programmatic**: `Pipeline(extractor) | transformer | loader`; chain with `|`; call `await pipeline.run()`.
- **Config-driven**: Load from explicit files (`from_config_files`), from a directory with `extract.yaml`, `transform.yaml`, `load.yaml` (`from_config_dir`), or from a single `pipeline.yaml` (`from_config_file`).
- **Variables**: Pass `PipelineContext(variables={"API_KEY": "x"})` or `variables={...}` in factory methods; `${VAR}` and `${VAR:-default}` in configs are resolved from these (no built-in `CONTRACT_DIR`).
- **Async**: `run()` is async; use `asyncio.run(pipeline.run())` in scripts or `await pipeline.run()` in async code.
- **Error handling**: Optional `error_context` with `ErrorMode` (STRICT, LENIENT, COLLECT) controls whether extraction/load failures raise or are collected in `result.errors`.

**Example**:
```python
import asyncio
from pycharter import Pipeline, PipelineContext, HTTPExtractor, PostgresLoader, Rename, AddField

# Programmatic
pipeline = (
    Pipeline(HTTPExtractor(url="https://api.example.com/users"))
    | Rename({"userName": "name"})
    | AddField("processed_at", "now()")
    | PostgresLoader(connection_string="...", table="users")
)
result = asyncio.run(pipeline.run())

# Config-driven (explicit files)
pipeline = Pipeline.from_config_files(
    extract="configs/extract.yaml",
    load="configs/load.yaml",
    variables={"API_KEY": "secret"}
)

# Config-driven (directory: extract.yaml, transform.yaml, load.yaml)
pipeline = Pipeline.from_config_dir("pipelines/users/")

# Config-driven (single file)
pipeline = Pipeline.from_config_file("pipelines/users/pipeline.yaml")

result = asyncio.run(pipeline.run())
```

**Exceptions**: Pipeline and config loading use PyCharter’s exception hierarchy: `PyCharterError` (base), `ConfigError`, `ConfigValidationError`, `ExpressionError`. See **Exceptions** under API Reference.

**See** `pycharter/etl_generator/ASYNC_AND_EXECUTION.md` for async usage and error modes.

---

### 6. 🔍 Quality Assurance (`pycharter.quality`)

**Purpose**: Data quality assurance pipeline that polices data according to data contracts, calculates quality metrics, tracks violations, and generates quality reports.

**When to Use**: When you need to:
- Monitor data quality over time
- Calculate quality scores and metrics
- Track and manage data quality violations
- Set quality thresholds and get alerts
- Generate quality reports for governance

**How It Works**:
- Validates data against contracts (using Runtime Validator)
- Calculates quality metrics (accuracy, completeness, violation rates)
- Tracks violations for audit and remediation
- Checks quality thresholds and generates alerts
- Produces comprehensive quality reports

**Example**:
```python
from pycharter import QualityCheck, QualityCheckOptions, QualityThresholds

# Define quality thresholds
thresholds = QualityThresholds(
    min_overall_score=95.0,
    max_violation_rate=0.05
)

# Run quality check
check = QualityCheck(store=store)
report = check.run(
    schema_id="user_schema_v1",
    data="data/users.json",
    options=QualityCheckOptions(
        calculate_metrics=True,
        record_violations=True,
        check_thresholds=True,
        thresholds=thresholds
    )
)

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

**Contribution to Journey**: The quality assurance module is the **policing layer** that ensures data quality is maintained throughout the pipeline. It provides metrics, tracking, and alerting capabilities that transform PyCharter from a contract management tool into a complete data quality assurance platform.

**See [Quality Module README](pycharter/quality/README.md) for detailed documentation.**

---

### Complete Workflow Example

Here's how all services work together in a complete data production journey:

```python
from pycharter import (
    parse_contract_file,
    SQLiteMetadataStore,
    from_dict,
    Validator,
    to_dict
)

# Step 1: Parse contract specification
metadata = parse_contract_file("user_contract.yaml")

# Step 2: Store metadata in database
store = SQLiteMetadataStore("metadata.db")
store.connect()
schema_id = store.store_schema("user", metadata.schema, version="1.0")

# Merge ownership and governance into metadata before storing
# Ownership and governance are part of metadata, not separate entities
metadata_dict = metadata.metadata.copy() if metadata.metadata else {}
if metadata.ownership:
    metadata_dict["business_owners"] = [metadata.ownership.get("owner", "unknown")] if metadata.ownership.get("owner") else []
if metadata.governance_rules:
    metadata_dict["governance_rules"] = metadata.governance_rules

# Store metadata once with all information (ownership and governance included)
store.store_metadata(resource_id=schema_id, resource_type="schema", metadata=metadata_dict)

# Store coercion and validation rules
store.store_coercion_rules(schema_id, {"age": "coerce_to_integer"}, version="1.0")
store.store_validation_rules(schema_id, {"age": {"is_positive": {}}}, version="1.0")

# Step 3: Generate Pydantic model from stored schema
schema = store.get_schema(schema_id)
UserModel = from_dict(schema, "User")

# Step 4: (Optional) Convert model back to schema for documentation
schema_doc = to_dict(UserModel)

# Step 5: Validate data in production pipeline
# Option A: Using Validator class (recommended for production)
validator = Validator(store=store, schema_id=schema_id)

def process_user_data(raw_data):
    result = validator.validate(raw_data)
    if result.is_valid:
        # Process validated data
        return result.data
    else:
        # Handle validation errors
        raise ValueError(f"Invalid data: {result.errors}")

# Option B: Using convenience function (quick start)
from pycharter import validate_with_store

def process_user_data_quick(raw_data):
    result = validate_with_store(store, schema_id, raw_data)
    if result.is_valid:
        return result.data
    else:
        raise ValueError(f"Invalid data: {result.errors}")
```

---

### 7. 🌐 REST API (`pycharter.api`)

**Purpose**: Expose all PyCharter services as REST API endpoints.

**When to Use**: When you need to use PyCharter from non-Python applications, microservices, or want to provide a web-based interface.

**How It Works**:
- Provides HTTP endpoints for all core services
- Uses FastAPI for automatic OpenAPI/Swagger documentation
- Supports both store-based and contract-based operations
- Handles request/response validation with Pydantic models
- Located at the root level (`api/`) as a separate application
- All endpoints are async-ready for better performance

**Example**:
```bash
# Start the API server (uses PYCHARTER_DATABASE_URL or sqlite:///pycharter.db)
pycharter api

# With host/port
pycharter api --host 0.0.0.0 --port 8080
```

**Endpoints**:
- `POST /api/v1/contracts/parse` - Parse a data contract
- `POST /api/v1/contracts/build` - Build contract from store
- `POST /api/v1/metadata/schemas` - Store a schema
- `GET /api/v1/metadata/schemas/{schema_id}` - Get a schema
- `POST /api/v1/schemas/generate` - Generate Pydantic model
- `POST /api/v1/validation/validate` - Validate data
- `POST /api/v1/validation/validate-batch` - Batch validation

**Documentation**: 
- Swagger UI: http://localhost:8000/docs
- ReDoc: http://localhost:8000/redoc

See `api/README.md` for complete API documentation.

### Service Integration Summary

| Service | Input | Output | Journey Stage |
|---------|-------|--------|---------------|
| **Contract Parser** | Contract files (YAML/JSON) | `ContractMetadata` | Contract Specification → Parsing |
| **Contract Builder** | Separate artifacts or Store | Consolidated contract | Storage → Consolidation |
| **Metadata Store** | `ContractMetadata` | Stored metadata (DB) | Parsing → Storage |
| **Pydantic Generator** | JSON Schema | Pydantic models | Storage → Model Generation |
| **JSON Schema Converter** | Pydantic models | JSON Schema | (Bidirectional) |
| **Runtime Validator** | Pydantic models + Data | `ValidationResult` | Model Generation → Validation |
| **ETL Pipelines** | Config files or code | `PipelineResult` | Extract → Transform → Load |
| **Quality Assurance** | Contract + Data | `QualityReport` | Validation → Quality Monitoring |

Each service is designed to be **independent** yet **composable**, allowing you to use them individually or together as part of a complete data contract management system.

## 📖 Documentation

- **[Configuration Guide](docs/guides/configuration.md)** - Database connection, `pycharter db init` / `upgrade` / `seed`, migrations, and variable injection
- **[Data Journey Guide](docs/guides/data-journey.md)** - Data production journey: