Quick Start¶

Get up and running with PyCharter in 5 minutes. The data contract is the single source of truth: it drives validation, quality checks, and ETL. See the Contract-First Workflow for the full define → parse/build → store → validate flow.

Quick Starts by Component¶

Component	What you do	Learn more
Data contracts	Define YAML/dict → parse or build → DataContract → validate / store / quality	Contract-First, Contract Parser, Contract Builder
Validator	Validate data against a contract (file, dict, DataContract, or store by name/version)	Validator API, Contracts Tutorial
ETL pipelines	Build pipelines with `Pipeline \\| extractor \\| transform \\| loader` or YAML config	Pipeline API, ETL Tutorial
Contract store	Store and retrieve contracts by contract name and contract version	Contract Store API, Contract Store Tutorial
Quality	Run quality checks with contract (file or name/version) and data; track violations	Quality API, Quality Tutorial
API & UI	Run `pycharter api` and `pycharter ui serve` for REST API and web UI	API & UI Tutorial

Your First Validation¶

The simplest way to use PyCharter is to validate data against a JSON schema:

from pycharter import from_dict, validate

# Define your schema
schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "name": {"type": "string", "minLength": 1},
        "email": {"type": "string", "format": "email"},
        "age": {"type": "integer", "minimum": 0}
    },
    "required": ["name", "email"]
}

# Generate a Pydantic model
User = from_dict(schema, "User")

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

print(f"Valid: {result.is_valid}")  # True
print(f"Data: {result.data}")       # User(name='Alice', email='alice@example.com', age=30)

Your First Data Contract¶

Define a contract (YAML or dict), parse it, promote to DataContract, and validate:

from pycharter import parse_contract_file, Validator

# Parse file → ContractMetadata → promote to DataContract
metadata = parse_contract_file("user_contract.yaml")
contract = metadata.to_data_contract()

# Validator accepts DataContract, dict, or file path
validator = Validator(contract_dict=contract)
result = validator.validate({"name": "Alice", "email": "alice@example.com"})
print(result.is_valid)  # True

Using the Validator Class¶

For production use, the Validator class caches the Pydantic model — fast for large batches. You can create it from a contract file, directory, DataContract, dict, or from the contract store (by contract name and version).

from pycharter import Validator

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

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

# From explicit files or dict
validator = Validator.from_files(schema="schemas/user.yaml", coercion_rules="rules/coerce.yaml")
validator = Validator.from_dict(schema={...}, coercion_rules={...}, validation_rules={...})

# From contract store (contract name and version)
validator = Validator(store=store, contract_name="user", contract_version="1.0.0")

# Validate one record
result = validator.validate({"name": "Alice", "email": "alice@example.com"})
print(result.is_valid)   # True
print(result.data)       # User(name='Alice', email='alice@example.com')
print(result.errors)     # [] when valid

# Invalid data
result = validator.validate({"name": "X"})  # missing required "email"
print(result.is_valid)   # False
print(result.errors)     # [{'loc': ('email',), 'msg': 'Field required', ...}]

# Batch validation (model cached across all calls)
results = validator.validate_batch([
    {"name": "Alice", "email": "alice@example.com"},
    {"name": "Bob", "email": "bob@example.com"},
    {"name": "Charlie", "email": "charlie@example.com"},
])
valid_count = sum(1 for r in results if r.is_valid)
print(f"Valid: {valid_count}/{len(results)}")

Using ValidatorBuilder (Fluent API)¶

ValidatorBuilder is a chainable interface for configuring validators with quality checks, state-specific rules, and metrics tracking:

from pycharter import ValidatorBuilder

validator = (
    ValidatorBuilder()
    .from_dir("contracts/order")
    .with_state_rules({
        # Per-lifecycle-state overrides: make fields optional in DRAFT, required in FILLED
        "DRAFT":  {"optional": ["filled_qty", "fill_price"]},
        "FILLED": {"required": ["filled_qty", "fill_price"]},
    })
    .with_quality_checks(thresholds={"completeness": 0.95, "accuracy": 0.99})
    .build()
)

# Standard validation
result = validator.validate({"symbol": "AAPL", "order_qty": 100})

# State-aware validation: rules change based on where the entity is in its lifecycle
draft_result  = validator.validate_for_state({"symbol": "AAPL", "order_qty": 100}, "DRAFT")
filled_result = validator.validate_for_state(
    {"symbol": "AAPL", "order_qty": 100, "filled_qty": 100, "fill_price": 150.0}, "FILLED"
)
print(draft_result.is_valid)   # True — filled_qty/fill_price are optional in DRAFT
print(filled_result.is_valid)  # True — filled_qty/fill_price are present and required in FILLED

validate_for_state(data, state) is especially powerful when combined with PyStator: use the FSM's current state to automatically apply the right validation rules.

Your First ETL Pipeline¶

Build a simple data pipeline:

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

# Define the pipeline
pipeline = (
    Pipeline(HTTPExtractor(url="https://jsonplaceholder.typicode.com/users"))
    | Rename({"username": "user_name"})  # Rename fields
    | Filter(lambda r: r.get("id", 0) <= 5)  # Keep first 5 users
    | FileLoader(path="output/users.json", file_format="json")
)

# Run the pipeline
result = asyncio.run(pipeline.run())

print(f"Extracted: {result.rows_extracted}")
print(f"Loaded: {result.rows_loaded}")
print(f"Duration: {result.duration_seconds:.2f}s")

Config-Driven Pipeline¶

Define pipelines in YAML for easier management:

extract.yamltransform.yamlload.yaml

type: http
url: https://jsonplaceholder.typicode.com/users
method: GET

rename:
  username: user_name
  email: contact_email

add:
  processed_at: "now()"

drop:
  - website
  - phone

type: file
file_path: output/users.json
format: json

Run the config-driven pipeline:

import asyncio
from pycharter import Pipeline

# Load from directory
pipeline = Pipeline.from_config_dir("pipelines/users/")

# Or from explicit files
pipeline = Pipeline.from_config_files(
    extract="configs/extract.yaml",
    transform="configs/transform.yaml",
    load="configs/load.yaml"
)

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

Data Quality Check¶

The fastest way to check data quality is check_quality() — one function, one result:

from pycharter import check_quality

report = check_quality(
    contract={"json_schema": {
        "version": "1.0.0",
        "properties": {
            "name": {"type": "string", "minLength": 1},
            "email": {"type": "string", "format": "email"},
            "age": {"type": "integer", "minimum": 0}
        },
        "required": ["name", "email"]
    }},
    data=[
        {"name": "Alice", "email": "alice@example.com", "age": 30},
        {"name": "Bob", "email": "invalid-email", "age": -5},
        {"name": "Charlie", "email": "charlie@example.com", "age": 25},
    ],
)

print(f"Quality Score: {report.quality_score.overall_score:.1f}/100")
print(f"Valid: {report.valid_count}/{report.record_count}")
print(f"Violations: {report.violation_count}")

Presets for Common Scenarios¶

Use presets instead of configuring every option:

from pycharter import check_quality, QualityCheckOptions

# Quick one-off check (default)
report = check_quality(contract, data)

# Strict gated check — must pass thresholds before proceeding
report = check_quality(contract, data, options=QualityCheckOptions.strict())
print(f"Passed thresholds: {report.passed}")

# Monitoring — dedup violations, skip unchanged data
report = check_quality(contract, data, options=QualityCheckOptions.monitoring())

Using the QualityCheck Class¶

For store-backed contracts or advanced control, use QualityCheck with contract name and version (or pass a contract file/dict/DataContract):

from pycharter import QualityCheck, QualityCheckOptions, QualityThresholds, PostgresContractStore

store = PostgresContractStore("postgresql://...")
store.connect()

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=90.0, max_violation_rate=0.10),
    ),
)

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

With an in-memory contract (file path, dict, or DataContract):

report = check.run(contract="user_contract.yaml", data=records, options=...)

Quick Data Profiling¶

Profile a dataset without a contract:

from pycharter import profile_data

profile = profile_data([
    {"name": "Alice", "age": 30},
    {"name": "Bob", "age": None},
])

print(f"Records: {profile['record_count']}")
print(f"Completeness: {profile['overall_stats']['completeness']:.1f}%")
print(f"Age nulls: {profile['field_profiles']['age']['null_count']}")

Domain Lifecycle & FSM Integration¶

PyCharter can validate that a data contract's states are aligned with an FSM definition (e.g. PyStator). This is useful to catch drift between your schema's enum values and your state machine's states.

from pycharter import check_state_alignment, validate_lifecycle_binding, get_lifecycle_binding

# Your FSM states (from PyStator or any other source)
fsm_states = {"PENDING", "OPEN", "FILLED", "CANCELLED"}

# Your contract (loaded from YAML or built programmatically)
contract = {
    "json_schema": {
        "properties": {
            "status": {"type": "string", "enum": ["PENDING", "OPEN", "FILLED"]}
        }
    },
    "metadata": {
        "governance_rules": {
            "lifecycle": {
                "state_machine_name": "order_management",
                "machine_version": "1.0.0",
                "state_field": "status",
                "entity_id_field": "order_id",
            }
        }
    }
}

# Check alignment: are contract enum values ↔ FSM states in sync?
result = check_state_alignment(contract, fsm_states, state_field="status")
print(result.aligned)                  # False — CANCELLED is in FSM but not in contract
print(result.missing_from_contract)    # {"CANCELLED"}
print(result.missing_from_fsm)         # set()

# Validate the lifecycle binding structure in metadata
errors = validate_lifecycle_binding(contract["metadata"])
print(errors)  # [] if the binding is structurally valid

# Read the binding
binding = get_lifecycle_binding(contract["metadata"])
print(binding.state_machine_name)  # "order_management"
print(binding.state_field)         # "status"

Start the API Server¶

Run PyCharter as a REST API:

# Start the API server
pycharter api

# Or with custom host/port
pycharter api --host 0.0.0.0 --port 8080

API endpoints:

Swagger UI: http://localhost:8002/docs
ReDoc: http://localhost:8002/redoc
OpenAPI JSON: http://localhost:8002/openapi.json

Example API call:

curl -X POST http://localhost:8002/api/v1/validation/validate \
  -H "Content-Type: application/json" \
  -d '{
    "json_schema": {
      "type": "object",
      "properties": {"name": {"type": "string"}},
      "required": ["name"]
    },
    "data": {"name": "Alice"}
  }'

Start the Web UI¶

Launch the interactive web interface:

# Production mode (pre-built)
pycharter ui serve

# Development mode (with hot reload)
pycharter ui dev

Open http://localhost:3000 to access:

Contract management
Schema browser
Validation testing
Quality dashboards
ETL monitoring

Next Steps¶

Now that you've seen the basics, dive deeper:

Contract-First Workflow — Define → parse/build → store → validate
Core Concepts — Data contracts, schemas, and the validation pipeline
ETL Pipelines Tutorial — Build production-ready data pipelines
Data Contracts Tutorial — Master validation and coercion
API Reference — Complete API documentation