Core services reference¶

PyCharter is organised as eight composable services around one contract artifact. This page is the deep tour: what each service does, where it sits in the data production journey, and how the pieces hand off to each other. The contract is what actually runs — the runtime validator is the enforcement layer, and quality assurance is the policing layer.

🏗️ 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:

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.json_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 contract store

Example:

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 contract store
contract = build_contract_from_store(store, "user_schema_v1")

# Use for validation - Validator merges rules internally
from pycharter import Validator

result = Validator(contract).validate({"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. 💾 Contract Store Client (`pycharter.contract_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: - PostgresContractStore - For PostgreSQL databases (recommended for production) - SQLiteContractStore - For SQLite databases (great for development and small deployments) - MongoDBContractStore - For MongoDB databases - RedisContractStore - For Redis databases - InMemoryContractStore - For testing and development (no persistence)

Example:

from pycharter import SQLiteContractStore, parse_contract_file

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

# Use SQLite contract store (or PostgresContractStore, MongoDBContractStore, RedisContractStore, etc.)
store = SQLiteContractStore("metadata.db")
store.connect()

# Store decomposed components
store.store_json_schema("user_schema", "1.0", metadata.json_schema)

# 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("user_schema", "1.0", metadata_dict)

# Store coercion and validation rules
store.store_coercion_rules(
    "user_schema", "1.0", {"rules": [{"field": "age", "function": "coerce_to_integer"}]}
)
store.store_validation_rules(
    "user_schema", "1.0", {"rules": [{"field": "age", "function": "is_positive"}]}
)

# Retrieve later
stored_schema = store.get_json_schema("user_schema", "1.0")
coercion_rules = store.get_coercion_rules("user_schema", "1.0")
validation_rules = store.get_validation_rules("user_schema", "1.0")

Contribution to Journey: The contract 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 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 contract store) - Programmatically generates Pydantic model classes at runtime - Enforces the common JSON Schema keywords plus custom coercions and validations; a Validator evaluates the remaining Draft 2020-12 assertions with a JSON Schema engine (see built-in coercions and validations) - Can generate models from dictionaries, JSON strings, files, or URLs - Optionally generates Python files with model definitions

Example:

from pycharter.pydantic_generator import from_dict
from pycharter import generate_model_file, ContractStoreClient

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

# Option 2: Generate from stored schema
client = ContractStoreClient(...)
schema = client.get_json_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:

from pycharter.json_schema_converter import to_dict
from pycharter import 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:

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
Tier 2: Convenience Functions (Quick start - model generation)
get_model_from_store() / get_model_from_contract() - Get a Pydantic model for reuse
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):

from pycharter import Validator, SQLiteContractStore

# 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 a contract dict (full artifact dict or bare JSON Schema)
validator = Validator.from_dict({"json_schema": {...}, "coercion_rules": {...}})
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 contract store (with database)
from pycharter.shared.refs import StoreRef

store = SQLiteContractStore("metadata.db")
store.connect()
validator = Validator.from_store(StoreRef("user_schema_v1", "1.0.0"), store=store)
result = validator.validate({"name": "Alice", "age": 30})

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

Example - Model Generation (Quick Start):

from pycharter import get_model_from_store, get_model_from_contract, SQLiteContractStore

# Generate a model from a store contract
store = SQLiteContractStore("metadata.db")
store.connect()
UserModel = get_model_from_store(store, "user_schema_v1", "1.0.0")

# Generate a model from a contract file (no database)
UserModel = get_model_from_contract("user_contract.yaml")

Example - Low-Level (When You Have a Model):

from pycharter.pydantic_generator import from_dict
from pycharter.runtime_validator import 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 contract stores) and contract-based workflows (for simpler use cases without database dependencies).

5b. 🔄 Pipelines (`pycharter.pipeline_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.from_dict({...}) with a steps: list; transformer objects still chain with | into a TransformerChain for standalone record processing. - Config-driven: Load from a directory containing pipeline.yaml (from_dir) or from an explicit path (from_file). Section-based multi-file configs (extract.yaml, transform.yaml, load.yaml) are no longer supported. - 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 sync (wraps asyncio.run); use await pipeline.arun() inside 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:

from pycharter import Pipeline

# Programmatic (dict form of the canonical steps)
pipeline = Pipeline.from_dict(
    {
        "name": "users_pipeline",
        "steps": [
            {
                "id": "fetch",
                "type": "extract",
                "source": {"type": "http", "url": "https://api.example.com/users"},
            },
            {
                "id": "shape",
                "type": "transform",
                "input": "fetch",
                "operations": [
                    {"rename": {"userName": "name"}},
                    {"add": {"processed_at": "now()"}},
                ],
            },
            {
                "id": "land",
                "type": "load",
                "input": "shape",
                "target": {
                    "type": "postgres",
                    "table": "users",
                    "mode": "upsert",
                    "match_on": "id",
                    "database": {"url": "postgresql://..."},
                },
            },
        ],
    }
)
result = pipeline.run()

# Config-driven (directory with pipeline.yaml)
pipeline = Pipeline.from_dir("pipelines/users/")

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

result = pipeline.run()

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

See pycharter/pipeline_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:

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 for detailed documentation.

Complete Workflow Example¶

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

from pycharter import (
    parse_contract_file,
    SQLiteContractStore,
    Validator,
)
from pycharter.json_schema_converter import to_dict
from pycharter.pydantic_generator import from_dict

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

# Step 2: Store metadata in database
store = SQLiteContractStore("metadata.db")
store.connect()
store.store_json_schema("user", "1.0", metadata.json_schema)

# 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("user", "1.0", metadata_dict)

# Store coercion and validation rules
store.store_coercion_rules(
    "user", "1.0", {"rules": [{"field": "age", "function": "coerce_to_integer"}]}
)
store.store_validation_rules(
    "user", "1.0", {"rules": [{"field": "age", "function": "is_positive"}]}
)

# Step 3: Generate Pydantic model from stored schema
schema = store.get_json_schema("user", "1.0")
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
from pycharter.shared.refs import StoreRef

validator = Validator.from_store(StoreRef("user", "1.0"), store=store)


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}")

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:

# 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 (see Swagger for full list): - Contracts: POST /api/v1/contracts/parse, POST /api/v1/contracts/build - Metadata: POST /api/v1/metadata/json_schemas, GET /api/v1/metadata/json_schemas/{contract_name}/{contract_version} - Validation: POST /api/v1/quality/validation/validate, POST /api/v1/quality/validation/validate-batch - Quality: POST /api/v1/quality/check, GET /api/v1/quality/metrics - ETL: POST /api/v1/pipeline/run (extract/transform/load YAML; optional pipeline quality checks) - Pipeline runs: GET /api/v1/pipeline/runs/legacy, GET /api/v1/pipeline/runs/legacy/stats, GET /api/v1/pipeline/registry - Vocabulary: GET /api/v1/semantic/schemes (legacy path), concepts, relationships, semantic health, and related endpoints

Documentation: - Swagger UI: http://localhost:8002/docs (default port 8002) - ReDoc: http://localhost:8002/redoc

See src/pycharter/api/README.md for complete API documentation.

8. 🌐 Web UI (`pycharter.ui`)¶

Purpose: Browser-based interface for contracts, pipelines, quality, and vocabulary. Served by pycharter ui serve (production) or pycharter ui dev (development with hot reload).

Main sections: - Contracts — Registry, diagram/form editor, and validation workspace - Pipelines — Overview dashboard, pipeline runs, ETL generator (YAML panels), and Pipeline diagram (Figma Jam–style visual editor for extract/transform/load nodes) - Quality — Quality metrics, thresholds, and violation tracking - Vocabulary — Concepts, concept schemes, diagrammatic workspace (ReactFlow), proposals, and semantic health - Documentation — In-app API playground and docs

Configure the API base URL in the UI Settings page. Authentication uses PYCHARTER_AUTH_USERS (see Configuration Guide).

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.

Core services reference¶

🏗️ Core Services & Data Production Journey¶

The Data Production Journey¶

1. 📄 Contract Parser (pycharter.contract_parser)¶

1b. 🏗️ Contract Builder (pycharter.contract_builder)¶

2. 💾 Contract Store Client (pycharter.contract_store)¶

3. 🏭 Pydantic Generator (pycharter.pydantic_generator)¶

4. 🔄 JSON Schema Converter (pycharter.json_schema_converter)¶

5. ✅ Runtime Validator (pycharter.runtime_validator)¶

5b. 🔄 Pipelines (pycharter.pipeline_generator)¶

6. 🔍 Quality Assurance (pycharter.quality)¶

Complete Workflow Example¶

7. 🌐 REST API (pycharter.api)¶

8. 🌐 Web UI (pycharter.ui)¶

Service Integration Summary¶

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

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

2. 💾 Contract Store Client (`pycharter.contract_store`)¶

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

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

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

5b. 🔄 Pipelines (`pycharter.pipeline_generator`)¶

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

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

8. 🌐 Web UI (`pycharter.ui`)¶