PyCharter Command & Function Reference¶

Quick reference for CLI commands and the Python API. PyCharter has three domains with three stores: Contract Store (contracts keyed by name and version), Pipeline Store (pipeline configs keyed by name and version), and Semantic Store (concept schemes, concepts, ontology configs). The data contract is the single source of truth for validation, quality, and pipelines.

CLI Commands¶

Database Management¶

Command	Description	Example Usage
`pycharter db init [database_url]`	Initialize database schema from scratch	`pycharter db init postgresql://user:pass@localhost/db`
`pycharter db upgrade [database_url]`	Upgrade database to latest revision	`pycharter db upgrade --revision head`
`pycharter db downgrade [database_url]`	Downgrade database to previous revision	`pycharter db downgrade --revision -1`
`pycharter db current [database_url]`	Show current database revision	`pycharter db current`
`pycharter db history [database_url]`	Show migration history	`pycharter db history`
`pycharter db stamp [revision] [database_url]`	Stamp database with revision without running migrations	`pycharter db stamp head`
`pycharter db seed [seed_dir] [database_url]`	Seed database from YAML files (default: bundled `pycharter/data/seed`)	`pycharter db seed` · `pycharter db seed ./my_seed postgresql://...`
`pycharter db truncate [database_url]`	Truncate all PyCharter tables (clear all data)	`pycharter db truncate --force`

API Server¶

Command	Description	Example Usage
`pycharter api`	Start the API server (default port 8002)	`pycharter api --host 0.0.0.0 --port 8002`
`pycharter api --reload`	Start API server with auto-reload	`pycharter api --reload`
`pycharter api --no-reload`	Start API server without auto-reload	`pycharter api --no-reload`

UI Server¶

Command	Description	Example Usage
`pycharter ui serve`	Serve the built UI (default port 3002)	`pycharter ui serve --port 3002`
`pycharter ui dev`	Run UI development server	`pycharter ui dev --api-url http://localhost:8002`
`pycharter ui build`	Build UI for production	`pycharter ui build`

Documentation (MkDocs)¶

Command	Description	Example Usage
`pycharter docs serve`	Serve docs locally (default: http://127.0.0.1:5002)	`pycharter docs serve --port 5002`
`pycharter docs build`	Build static site to `site/`	`pycharter docs build`

Validation Worker¶

Command	Description	Example Usage
`pycharter worker start`	Run the validation worker (polls DB for jobs)	`pycharter worker start`
`pycharter worker start --concurrency N`	Set number of poller tasks	`pycharter worker start --concurrency 5`
`pycharter worker start --poll-interval MS`	Poll interval in milliseconds	`pycharter worker start --poll-interval 1000`

Quality Assurance¶

Command	Description	Example Usage
`pycharter quality check --contract-name X --contract-version Y --data <file>`	Run quality check using store (requires `--database-url`)	`pycharter quality check --contract-name user --contract-version 1.0.0 --data data.json --database-url postgresql://...`
`pycharter quality check --contract <file> --data <file>`	Run quality check with contract file	`pycharter quality check --contract contract.yaml --data data.json`
`pycharter quality violations`	List recorded violations (optional `--contract-name`, `--contract-version`, `--database-url`)	`pycharter quality violations --contract-name user --contract-version 1.0.0 --status open`

Python Functions & Classes¶

Contract Parser¶

Function/Class	Description	Example Usage
`parse_contract_file(path)`	Parse contract from YAML/JSON file	`metadata = parse_contract_file("contract.yaml")`
`parse_contract(dict)`	Parse contract from dictionary	`metadata = parse_contract({"json_schema": {...}})`
`ContractMetadata`	Parsed contract metadata object	`metadata.schema`, `metadata.governance_rules`, `metadata.ontology`

Contract Builder¶

Function/Class	Description	Example Usage
`build_contract(artifacts)`	Build full contract from artifacts; returns DataContract	`contract = build_contract(ContractArtifacts(...))`
`build_contract_from_store(store, contract_name, contract_version, ...)`	Rebuild contract from contract store; returns DataContract	`contract = build_contract_from_store(store, "user", "1.0.0")`
`DataContract`	First-class contract type (json_schema, coercion_rules, validation_rules, metadata, field_mapping, versions); use `.to_dict()` for dict form	`contract.json_schema`, `contract.to_dict()`
`ContractArtifacts`	Container for contract artifacts	`artifacts = ContractArtifacts(schema={...}, metadata={...}, ...)`

Contract Store¶

All store methods are keyed by contract_name and contract_version. Use ContractStoreClient implementations (e.g. PostgresContractStore, MongoDBContractStore).

Function/Class	Description	Example Usage
`PostgresContractStore(connection_string)`	PostgreSQL contract store	`store = PostgresContractStore("postgresql://...")`
`MongoDBContractStore(connection_string, database_name)`	MongoDB contract store	`store = MongoDBContractStore("mongodb://...", "db")`
`InMemoryContractStore()`	In-memory contract store	`store = InMemoryContractStore()`
`RedisContractStore(connection_string)`	Redis contract store	`store = RedisContractStore("redis://...")`
`store.connect()`	Connect to contract store	`store.connect()`
`store.disconnect()`	Disconnect from contract store	`store.disconnect()`
`store.store_schema(contract_name, contract_version, schema)`	Store JSON Schema for contract	`store.store_schema("user", "1.0.0", schema)`
`store.get_schema(contract_name, contract_version)`	Retrieve schema for contract	`schema = store.get_schema("user", "1.0.0")`
`store.store_metadata(contract_name, contract_version, metadata)`	Store metadata for contract	`store.store_metadata("user", "1.0.0", metadata_dict)`
`store.get_metadata(contract_name, contract_version)`	Retrieve metadata for contract	`metadata = store.get_metadata("user", "1.0.0")`
`store.store_coercion_rules(contract_name, contract_version, rules)`	Store coercion rules	`store.store_coercion_rules("user", "1.0.0", rules)`
`store.get_coercion_rules(contract_name, contract_version)`	Retrieve coercion rules	`rules = store.get_coercion_rules("user", "1.0.0")`
`store.store_validation_rules(contract_name, contract_version, rules)`	Store validation rules	`store.store_validation_rules("user", "1.0.0", rules)`
`store.get_validation_rules(contract_name, contract_version)`	Retrieve validation rules	`rules = store.get_validation_rules("user", "1.0.0")`
`store.get_contract(contract_name, contract_version, ...)`	Retrieve full contract dict	`contract = store.get_contract("user", "1.0.0")`
`store.list_contracts()`	List all stored contracts	`contracts = store.list_contracts()`
`store.store_field_mapping(contract_name, contract_version, field_mapping)`	Store field mapping	`store.store_field_mapping("user", "1.0.0", mapping)`
`store.get_field_mapping(contract_name, contract_version)`	Retrieve field mapping	`mapping = store.get_field_mapping("user", "1.0.0")`

Pipeline Store¶

Pipeline configs are keyed by pipeline_name and pipeline_version. Use PipelineStoreClient implementations (e.g. PostgresPipelineStore, SQLitePipelineStore). See Pipeline Store.

Function/Class	Description	Example Usage
`PostgresPipelineStore(connection_string)`	PostgreSQL pipeline store	`store = PostgresPipelineStore("postgresql://...")`
`SQLitePipelineStore(connection_string)`	SQLite pipeline store	`store = SQLitePipelineStore("sqlite:///pycharter.db")`
`InMemoryPipelineStore()`	In-memory pipeline store	`store = InMemoryPipelineStore()`
`store.store_config(pipeline_name, pipeline_version, steps, ...)`	Store pipeline config	`store.store_config("daily_orders", "1.0.0", steps, settings=...)`
`store.get_config(pipeline_name, pipeline_version)`	Retrieve full config	`config = store.get_config("daily_orders", "1.0.0")`
`store.get_pipeline_dict(pipeline_name, pipeline_version)`	Config for `Pipeline.from_dict()`	`d = store.get_pipeline_dict("daily_orders", "1.0.0")`
`store.list_configs(pipeline_name=None)`	List configs	`store.list_configs()`
`store.list_versions(pipeline_name)`	List versions	`store.list_versions("daily_orders")`

Semantic Store¶

Concept schemes, concepts, and ontology configs. Use SemanticStoreClient implementations (e.g. PostgresSemanticStore, SQLiteSemanticStore). See Semantic Store.

Function/Class	Description	Example Usage
`PostgresSemanticStore(connection_string)`	PostgreSQL semantic store	`store = PostgresSemanticStore("postgresql://...")`
`SQLiteSemanticStore(connection_string)`	SQLite semantic store	`store = SQLiteSemanticStore("sqlite:///pycharter.db")`
`InMemorySemanticStore()`	In-memory semantic store	`store = InMemorySemanticStore()`
`store.store_scheme(scheme_key, version, title, ...)`	Store concept scheme	`store.store_scheme("trading-domain", "1.0.0", "Trading Domain")`
`store.get_scheme(scheme_key, version=None)`	Retrieve scheme	`store.get_scheme("trading-domain")`
`store.store_concept(name, scheme_key, ...)`	Store concept	`store.store_concept("OrderId", "trading-domain", label="Order ID")`
`store.get_concept(name)`	Retrieve concept	`store.get_concept("OrderId")`
`store.list_concepts(scheme_key=None)`	List concepts	`store.list_concepts("trading-domain")`
`store.store_relationship(source, target, rel_type)`	Store relationship	`store.store_relationship("OrderId", "CustomerId", "references")`
`store.store_config(ontology_name, ontology_version, linkml_yaml, ...)`	Store ontology config	`store.store_config("trading", "1.0.0", yaml_str)`
`store.get_config(ontology_name, ontology_version)`	Retrieve ontology config	`store.get_config("trading", "1.0.0")`

Pydantic Generator¶

Function/Class	Description	Example Usage
`from_dict(schema_dict, model_name)`	Generate Pydantic model from dict	`UserModel = from_dict(schema, "User")`
`from_file(file_path, model_name)`	Generate Pydantic model from file	`UserModel = from_file("schema.json", "User")`
`from_json(json_string, model_name)`	Generate Pydantic model from JSON string	`UserModel = from_json(json_str, "User")`
`from_url(url, model_name)`	Generate Pydantic model from URL	`UserModel = from_url("https://...", "User")`
`generate_model(schema, model_name)`	Generate Pydantic model (generic)	`UserModel = generate_model(schema, "User")`
`generate_model_file(schema, output_path, model_name)`	Generate and save model to file	`generate_model_file(schema, "models.py", "User")`

JSON Schema Converter¶

Function/Class	Description	Example Usage
`to_dict(model)`	Convert Pydantic model to dict	`schema = to_dict(UserModel)`
`to_file(model, file_path)`	Convert model to file	`to_file(UserModel, "schema.json")`
`to_json(model)`	Convert model to JSON string	`json_str = to_json(UserModel)`
`model_to_schema(model)`	Convert model to JSON Schema	`schema = model_to_schema(UserModel)`

Runtime Validator¶

Function/Class	Description	Example Usage
`validate(model, data)`	Validate single record	`result = validate(UserModel, data)`
`validate_batch(model, data_list)`	Validate batch of records	`results = validate_batch(UserModel, data_list)`
`validate_with_store(store, contract_name, contract_version, data)`	Validate using store	`result = validate_with_store(store, "user", "1.0.0", data)`
`validate_batch_with_store(store, contract_name, contract_version, data_list)`	Batch validate using store	`results = validate_batch_with_store(store, "user", "1.0.0", data_list)`
`validate_with_contract(contract, data)`	Validate using contract (dict or DataContract)	`result = validate_with_contract(contract, data)`
`validate_batch_with_contract(contract, data_list)`	Batch validate using contract	`results = validate_batch_with_contract(contract, data_list)`
`get_model_from_store(store, contract_name, contract_version, model_name)`	Get model from store	`UserModel = get_model_from_store(store, "user", "1.0.0", "User")`
`get_model_from_contract(contract, model_name)`	Get model from contract (dict or DataContract)	`UserModel = get_model_from_contract(contract, "User")`
`ValidationResult`	Validation result object	`result.is_valid`, `result.errors`, `result.data`
`validate_stream(model, data_stream)`	Validate streaming data	`results = validate_stream(UserModel, stream)`
`validate_async(model, data)`	Async validation	`result = await validate_async(UserModel, data)`
`validate_batch_async(model, data_list)`	Async batch validation	`results = await validate_batch_async(UserModel, data_list)`
`StreamingValidator`	Streaming validator with callbacks and statistics	`validator = StreamingValidator(model, on_valid=..., on_invalid=...)`

Quality Assurance¶

Convenience functions¶

Function	Description	Example Usage
`check_quality(contract, data, options)`	One-liner quality check	`report = check_quality(contract, data)`
`check_quality_with_store(store, name, version, data)`	Quality check via contract store	`report = check_quality_with_store(store, "user", "1.0.0", data)`
`profile_data(data, fields)`	Standalone data profiling	`profile = profile_data(records)`

Presets¶

Preset	Description	Example
`QualityCheckOptions.basic()`	Metrics + violations, no profiling/thresholds	Default for `check_quality()`
`QualityCheckOptions.strict()`	Everything on, including default thresholds	`check_quality(c, d, QualityCheckOptions.strict())`
`QualityCheckOptions.monitoring()`	Strict + skip-if-unchanged + dedup	For scheduled/recurring checks

Classes¶

Function/Class	Description	Example Usage
`QualityCheck(store, db_session)`	Quality check orchestrator	`check = QualityCheck(store=store)`
`check.run(contract_name, contract_version, contract, data, options)`	Run quality check (use contract_name+contract_version for store, or contract for file/dict/DataContract)	`report = check.run(contract_name="user", contract_version="1.0.0", data=data, options=...)`
`QualityCheckOptions`	Quality check configuration	`options = QualityCheckOptions(calculate_metrics=True)`
`QualityReport`	Quality check report	`report.quality_score`, `report.violation_count`, `report.schema_id` (e.g. "name:version")
`QualityScore`	Quality metrics	`score.overall_score`, `score.accuracy`, `score.completeness`
`QualityThresholds`	Quality thresholds for alerting	`thresholds = QualityThresholds(min_overall_score=95.0)`
`QualityMetrics`	Quality metrics calculator	`metrics = QualityMetrics()`
`ViolationTracker`	Violation tracking system	`tracker = ViolationTracker(db_session=session)`
`tracker.record_violation(...)`	Record a violation	`tracker.record_violation(schema_id, record_id, data, result)` (schema_id is "name:version")
`tracker.get_violations(...)`	Query violations	`violations = tracker.get_violations(schema_id="user:1.0.0")`
`tracker.resolve_violation(violation_id, resolved_by)`	Resolve a violation	`tracker.resolve_violation(violation_id, "user@example.com")`
`ViolationRecord`	Individual violation record	`violation.field_name`, `violation.error_message`
`DataProfiler`	Data profiling tool	`profiler = DataProfiler()`
`profiler.profile(data, fields)`	Profile dataset	`profile = profiler.profile(data_list)`
`FieldQualityMetrics`	Per-field quality metrics	`metrics.field_name`, `metrics.completeness`

QualityCheckOptions Parameters:

record_violations: Record violations to database (default: True)
calculate_metrics: Calculate quality metrics (default: True)
check_thresholds: Check against quality thresholds (default: False)
thresholds: QualityThresholds object for threshold checking (optional)
include_profiling: Include data profiling in report (default: False)
include_field_metrics: Include per-field metrics (default: True)
sample_size: Process only a sample of records (optional)
metadata: Additional metadata dictionary (optional)
data_version: Version identifier for dataset (optional)
data_source: Source identifier - file path, table name, etc. (optional)
skip_if_unchanged: Skip check if data hasn't changed (requires data fingerprint, default: False)
deduplicate_violations: Deduplicate violations for same record+field+error (default: True)

Pipeline Quality (Post-Load Checks)¶

Class/Alias	Description	Example Usage
`PostLoadChecker` / `QualityChecker`	Column/dataset checks after ETL load	`checker = PostLoadChecker(checks, pipeline_name="orders")`

DLQ (Dead Letter Queue)¶

Function	Description	Example Usage
`add_record_sync(dlq, ...)`	Sync wrapper for `dlq.add_record()`	`record = add_record_sync(dlq, "pipeline", data, reason, ...)`
`add_batch_sync(dlq, ...)`	Sync wrapper for `dlq.add_batch()`	`records = add_batch_sync(dlq, "pipeline", batch, reason, ...)`
`retry_record_sync(dlq, record_id)`	Sync wrapper for `dlq.retry_record()`	`ok = retry_record_sync(dlq, "abc123")`

Quick Examples¶

Complete Workflow¶

from pycharter import (
    parse_contract_file,
    build_contract_from_store,
    PostgresContractStore,
    from_dict,
    validate,
    QualityCheck,
    QualityCheckOptions,
)

# 1. Parse contract and get DataContract (or build from artifacts)
metadata = parse_contract_file("contract.yaml")
contract = metadata.to_data_contract()

# 2. Store in database (keyed by contract name and version)
store = PostgresContractStore("postgresql://...")
store.connect()
store.store_schema("user", "1.0.0", contract.json_schema)
store.store_metadata("user", "1.0.0", contract.metadata or {})
store.store_coercion_rules("user", "1.0.0", contract.coercion_rules or {})
store.store_validation_rules("user", "1.0.0", contract.validation_rules or {})

# 3. Rebuild contract from store (optional)
contract = build_contract_from_store(store, "user", "1.0.0")
schema = store.get_schema("user", "1.0.0")
UserModel = from_dict(schema, "User")

# 4. Validate data
result = validate(UserModel, data)
if result.is_valid:
    print("Valid!")

# 5. Quality check (by contract name and version, or pass contract)
check = QualityCheck(store=store)
report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data="data.json",
    options=QualityCheckOptions(
        calculate_metrics=True,
        record_violations=True,
        include_profiling=True,
        data_version="v1.0.0",
        data_source="data.json",
        deduplicate_violations=True,
        skip_if_unchanged=True,
    ),
)
print(f"Quality Score: {report.quality_score.overall_score}")
store.disconnect()

CLI Workflow¶

# Initialize database
pycharter db init postgresql://user:pass@localhost/db

# Run quality check (contract name + version, requires --database-url)
pycharter quality check --contract-name user --contract-version 1.0.0 --data data.json --database-url postgresql://...

# Or with a contract file
pycharter quality check --contract contract.yaml --data data.json

# List violations
pycharter quality violations --contract-name user --contract-version 1.0.0 --database-url postgresql://...

# Start API server
pycharter api --port 8002

# Start UI
pycharter ui serve --port 3002

Notes¶

All database commands support PYCHARTER_DATABASE_URL environment variable
Quality checks can persist to database when db_session is provided
Metadata stores support PostgreSQL, MongoDB, Redis, and InMemory
All validation functions return ValidationResult objects
Quality reports include metrics, violations, and optional profiling data
Violation Deduplication: Violations are automatically deduplicated (same record+field+error only recorded once)
Data Fingerprinting: Quality checks automatically calculate data fingerprints to detect unchanged data
Version Tracking: Use data_version and data_source options to track data versions and sources
Skip Unchanged: Set skip_if_unchanged=True to avoid creating duplicate metrics for unchanged data