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).

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 key: json_schema; "schema" also accepted)
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]

metadata:
  title: User Contract
  description: User records for the platform
  version: "1.0.0"

ownership:
  owner: data-team
  steward: alice@example.com

You can add coercion rules, validation rules, and field mapping (e.g. ontology bindings) in the same file or via the store.

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 (use json_schema; "schema" also accepted)
metadata = parse_contract({"json_schema": {...}, "metadata": {...}})
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(
    schema=metadata.schema,
    metadata=metadata.metadata or {},
    coercion_rules=metadata.coercion_rules or {},
    validation_rules=metadata.validation_rules or {},
    # optional: ownership, governance_rules, field_mapping
)
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 mapping
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 {})
if contract.field_mapping:
    store.store_field_mapping("user", "1.0.0", contract.field_mapping)

# 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, validate_with_contract

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

# Or one-liner with contract
result = validate_with_contract(contract, data)

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

from pycharter import Validator

validator = Validator(store=store, contract_name="user", contract_version="1.0.0")
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.

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.