Quick Start (component tour)

New users

For the shortest path to a working example, see Start here and the Cookbook. This page tours all major components (~20 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 Validator

# 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"],
}

# Build a validator straight from the schema
validator = Validator.from_dict(schema)

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

print(f"Valid: {result.is_valid}")  # True
print(f"Data: {result.data}")

# Failures come back as human-readable diagnostics
result = validator.validate({"name": "Bob", "email": "bob@x.io", "age": "N/A"})
print(result.errors[0])  # Column 'age': expected integer, found string 'N/A'

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.from_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 a contract dict (full artifact dict or a bare JSON Schema document)
validator = Validator.from_dict(
    {
        "json_schema": {...},
        "coercion_rules": {...},
        "validation_rules": {...},
    }
)

# From contract store (contract name and version)
from pycharter.shared.refs import StoreRef

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

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

Configuring a Validator

Configuration options are keyword-only arguments on the constructor and the from_* classmethods — quality checks, state-specific rules, strict mode, and metrics tracking:

from pycharter import Validator

validator = Validator.from_dir(
    "contracts/order",
    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"]},
    },
    include_quality=True,
)

# 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 pycharter

result = pycharter.run(
    {
        "name": "users_pipeline",
        "steps": [
            {
                "id": "fetch_users",
                "type": "extract",
                "source": {
                    "type": "http",
                    "url": "https://jsonplaceholder.typicode.com/users",
                },
            },
            {
                "id": "shape_users",
                "type": "transform",
                "input": "fetch_users",
                "operations": [{"rename": {"username": "user_name"}}],
            },
            {
                "id": "save_users",
                "type": "load",
                "input": "shape_users",
                "target": {
                    "type": "file",
                    "path": "output/users.json",
                    "format": "json",
                    "mode": "overwrite",
                },
            },
        ],
    }
)

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 a single canonical pipeline.yaml:

pipeline.yaml

api_version: pycharter.io/v2
kind: Pipeline
metadata:
  name: users
  namespace: examples
  version: "1.0.0"
  title: Users

spec:
  variables:
    OUTPUT_PATH: output/users.json

  steps:
    - id: extract_users
      type: extract
      source:
        type: http
        url: https://jsonplaceholder.typicode.com/users
        method: GET

    - id: transform_users
      type: transform
      input: extract_users
      operations:
        - type: rename
          mapping:
            username: user_name
            email: contact_email
        - type: add
          fields:
            processed_at: "now()"
        - type: drop
          fields: [website, phone]

    - id: load_users
      type: load
      input: transform_users
      target:
        type: file
        path: ${OUTPUT_PATH}
        format: json

Run the config-driven pipeline:

from pycharter import Pipeline

pipeline = Pipeline.from_dir("pipelines/users/")
# Or: Pipeline.from_file("pipelines/users/pipeline.yaml")

result = 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/quality/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:3002 to access:

• Contract management
• Schema browser
• Validation testing
• Quality dashboards
• ETL monitoring

More usage examples

Basic Usage

Using Convenience Functions (Quick Start):

from pycharter.pydantic_generator import from_dict, from_file
from pycharter import from_json

# From dictionary
schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "title": {"type": "string"},
        "published": {"type": "boolean", "default": False},
    },
}
Article = from_dict(schema, "Article")

# From JSON string
schema_json = (
    '{"type": "object", "version": "1.0.0", "properties": {"name": {"type": "string"}}}'
)
User = from_json(schema_json, "User")

# From file
Product = from_file("product_schema.json", "Product")

Using Validator Class (Production):

from pycharter import Validator

# From directory or single file
validator = Validator.from_dir("data/contracts/article")
# or: validator = Validator.from_file("article_contract.yaml")
result = validator.validate({"title": "My Article", "published": True})

Nested Objects

from pycharter.pydantic_generator import from_dict

schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "name": {"type": "string"},
        "address": {
            "type": "object",
            "properties": {
                "street": {"type": "string"},
                "city": {"type": "string"},
                "zipcode": {"type": "string"},
            },
        },
    },
}

Person = from_dict(schema, "Person")
person = Person(
    name="Alice",
    address={"street": "123 Main St", "city": "New York", "zipcode": "10001"},
)

print(person.address.city)  # Output: New York

Arrays and Collections

from pycharter.pydantic_generator import from_dict

schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "tags": {"type": "array", "items": {"type": "string"}},
        "items": {
            "type": "array",
            "items": {
                "type": "object",
                "properties": {"name": {"type": "string"}, "price": {"type": "number"}},
            },
        },
    },
}

Cart = from_dict(schema, "Cart")
cart = Cart(
    tags=["python", "pydantic"],
    items=[{"name": "Apple", "price": 1.50}, {"name": "Banana", "price": 0.75}],
)

print(cart.items[0].name)  # Output: Apple

Coercion and Validation

PyCharter supports coercion (pre-validation transformation) and validation (post-validation checks):

from pycharter.pydantic_generator import from_dict

schema = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "flight_number": {
            "type": "integer",
            "coercion": "coerce_to_integer",  # Convert string/float to int
        },
        "destination": {
            "type": "string",
            "coercion": "coerce_to_string",
            "validations": {
                "min_length": {"threshold": 3},
                "max_length": {"threshold": 3},
                "no_capital_characters": None,
                "only_allow": {"allowed_values": ["abc", "def", "ghi"]},
            },
        },
        "distance": {
            "type": "number",
            "coercion": "coerce_to_float",
            "validations": {"greater_than_or_equal_to": {"threshold": 0}},
        },
    },
}

Flight = from_dict(schema, "Flight")

# Coercion happens automatically
flight = Flight(
    flight_number="123",  # Coerced to int: 123
    destination="abc",  # Passes all validations
    distance="100.5",  # Coerced to float: 100.5
)

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