Contract-First Workflow

The data contract is the single source of truth for structure, validation, pipelines, and quality in PyCharter. This guide walks through a contract-first workflow: define → parse/build → store → validate (and optionally ETL + quality).

Choosing between PyCharter and tools like Great Expectations or Soda? See PyCharter vs. the alternatives.

1. Define the contract

Define your contract as YAML (or a dict) with schema and optional coercion, validation, metadata, and field mapping:

# user_contract.yaml (canonical sections nest under spec)
api_version: pycharter.io/v2
kind: DataContract
metadata:
  name: user_contract
  version: "1.0.0"
  description: User records for the platform
  ownership:
    roles:
      business_owner: [data-team]
      data_steward: [alice@example.com]
spec:
  json_schema:
    type: object
    version: "1.0.0"
    properties:
      name: { type: string, minLength: 1 }
      email: { type: string, format: email }
      age: { type: integer }
    required: [name, email]

You can add coercion rules, validation rules, and field bindings (links from contract fields to vocabulary concepts) in the same file or via the store.

Prefer starting from shipped artifacts? See Templates overview (template_contract.yaml, REST /ops/templates/contract-artifacts, or pycharter contract sketch).

When you run pipelines or streams, continue to Contract → ETL → quality for step contract: flags and quality paths.

2. Parse or build the contract

From file or dict: use the contract parser to get a ContractMetadata object, then promote it to the canonical DataContract type:

from pycharter import parse_contract_file, parse_contract

# From file
metadata = parse_contract_file("user_contract.yaml")
contract = metadata.to_data_contract()  # DataContract

# From dict (manifest envelope: sections nest under spec)
metadata = parse_contract(
    {
        "api_version": "pycharter.io/v2",
        "kind": "DataContract",
        "metadata": {"name": "user", "version": "1.0.0"},
        "spec": {"json_schema": {...}},
    }
)
contract = metadata.to_data_contract()

From artifacts: use the contract builder to produce a DataContract from ContractArtifacts:

from pycharter import build_contract, ContractArtifacts, DataContract

artifacts = ContractArtifacts(
    json_schema=metadata.json_schema,
    metadata=metadata.metadata or {},
    coercion_rules=metadata.coercion_rules or {},
    validation_rules=metadata.validation_rules or {},
    # optional: ownership, governance_rules, field_bindings
)
contract = build_contract(artifacts)  # returns DataContract

From the store: if the contract is already stored, build it by name and version:

from pycharter import build_contract_from_store, PostgresContractStore

store = PostgresContractStore("postgresql://...")
store.connect()
contract = build_contract_from_store(store, "user", "1.0.0")  # DataContract
store.disconnect()

3. Store by contract name and version

Use ContractStoreClient to store and retrieve contracts keyed by contract name and contract version:

from pycharter import PostgresContractStore

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

# Store schema, metadata, coercion/validation rules, and optional field bindings
store.store_json_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 {})
if contract.field_bindings:
    store.store_field_bindings("user", "1.0.0", contract.field_bindings)

# List all contracts
contracts = store.list_contracts()  # [{"name": "user", "version": "1.0.0"}, ...]

# Retrieve full contract
full = store.get_contract("user", "1.0.0")

store.disconnect()

4. Use the contract everywhere

Validation: Pass the contract (or its dict form) to the validator:

from pycharter import Validator

# Validator accepts DataContract, dict, or ContractMetadata
validator = Validator(contract)  # or Validator.from_dict(contract.to_dict())
result = validator.validate({"name": "Alice", "email": "alice@example.com", "age": 30})

Store-based validation: Use contract name and version with a connected store:

from pycharter import Validator
from pycharter.shared.refs import StoreRef

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

Quality checks: Run quality checks with either a contract file/dict or contract name and version (and a database for violations/metrics):

from pycharter import QualityCheck, QualityCheckOptions

check = QualityCheck(store=store)
report = check.run(
    contract_name="user",
    contract_version="1.0.0",
    data=records,
    options=QualityCheckOptions(record_violations=True),
)
# Or with in-memory contract:
report = check.run(contract=contract, data=records, options=...)

CLI:

# Quality check using contract name and version (requires --database-url)
pycharter quality check --contract-name user --contract-version 1.0.0 --data data.json --database-url postgresql://...

# Quality check using a contract file
pycharter quality check --contract user_contract.yaml --data data.json

# List violations (optional filters by contract name/version)
pycharter quality violations --contract-name user --contract-version 1.0.0 --database-url postgresql://...

ETL: Reference the contract in your ETL config (e.g. by contract name/version); load validation uses the same contract for structure and rules.

Aggregate quality (optional): Add a top-level data_quality section with rules and enable it per pipeline step with quality: true. See Contract vs pipeline quality and Periodic quality monitoring.

Quick score without a stored contract:

from pycharter import check_quality

report = check_quality(contract=contract, data=records)
print(report.quality_score.overall_score, report.valid_count, report.invalid_count)

Summary flow

Define (YAML/dict) → Parse/build → DataContract → Store (name + version)
                                                         ↓
Validate / ETL / Quality ←───────────────────────────────┘

Use DataContract (or its dict form) consistently for validation, quality, and ETL so structure, rules, and metadata stay aligned.