Skip to content

Contract Store & Schema Registry

Learn to store, version, and manage your data contracts in a centralized registry.

What You'll Learn

  • Set up different contract store backends
  • Store and version schemas
  • Manage coercion and validation rules
  • Track ownership and governance
  • Query and retrieve metadata
  • Handle schema evolution

Prerequisites

pip install pycharter

For specific backends: 

pip install psycopg2-binary  # PostgreSQL
pip install pymongo          # MongoDB
pip install redis            # Redis

Part 1: Store Backends

PyCharter supports multiple storage backends:

Backend Best For
InMemoryContractStore Testing, development
SQLiteContractStore Single-user, local dev
PostgresContractStore Production, multi-user
MongoDBContractStore Document-oriented
RedisContractStore High-performance caching

In-Memory Store

from pycharter import InMemoryContractStore

store = InMemoryContractStore()
store.connect()

# Data is lost when the program exits

SQLite Store

from pycharter import SQLiteContractStore

store = SQLiteContractStore("metadata.db")
store.connect()

# Database file persists between runs

PostgreSQL Store

from pycharter import PostgresContractStore

store = PostgresContractStore(
    "postgresql://user:password@localhost:5432/pycharter"
)
store.connect()

# Production-ready with full ACID compliance

MongoDB Store

from pycharter import MongoDBContractStore

store = MongoDBContractStore(
    "mongodb://localhost:27017/pycharter"
)
store.connect()

# Document-oriented storage

Redis Store

from pycharter import RedisContractStore

store = RedisContractStore("redis://localhost:6379/0")
store.connect()

# High-performance key-value storage

Part 2: Storing Schemas

Basic Schema Storage

from pycharter import SQLiteContractStore

store = SQLiteContractStore("metadata.db")
store.connect()

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

# Store the schema (keyed by contract name and version)
store.store_schema("user", "1.0.0", schema)
print("Stored contract user @ 1.0.0")

Schema Versioning

# Store version 1.0.0
schema_v1 = {
    "type": "object",
    "version": "1.0.0",
    "properties": {
        "name": {"type": "string"},
        "email": {"type": "string"}
    }
}
store.store_schema("user", "1.0.0", schema_v1)

# Store version 2.0.0 (added age field)
schema_v2 = {
    "type": "object",
    "version": "2.0.0",
    "properties": {
        "name": {"type": "string"},
        "email": {"type": "string"},
        "age": {"type": "integer"}  # New field
    }
}
store.store_schema("user", "2.0.0", schema_v2)

# Retrieve by contract name and version
schema = store.get_schema("user", "1.0.0")
schema_v2_retrieved = store.get_schema("user", "2.0.0")

Listing Contracts

# List all stored contracts
contracts = store.list_contracts()
for c in contracts:
    print(f"{c['name']} @ {c['version']}")

Part 3: Coercion and Validation Rules

Storing Rules

# Store coercion rules
coercion_rules = {
    "version": "1.0.0",
    "rules": {
        "age": "coerce_to_integer",
        "email": "coerce_to_lowercase",
        "created_at": "coerce_to_datetime"
    }
}

store.store_coercion_rules("user", "1.0.0", coercion_rules)

# Store validation rules
validation_rules = {
    "version": "1.0.0",
    "rules": {
        "age": {
            "is_positive": {},
            "less_than_or_equal_to": {"threshold": 150}
        },
        "email": {
            "is_email": {}
        },
        "name": {
            "min_length": {"threshold": 1},
            "max_length": {"threshold": 100}
        }
    }
}

store.store_validation_rules("user", "1.0.0", validation_rules)

Retrieving Rules

# Get coercion and validation rules (by contract name and version)
coercion = store.get_coercion_rules("user", "1.0.0")
validation = store.get_validation_rules("user", "1.0.0")
print(f"Coercion rules: {coercion}")
print(f"Validation rules: {validation}")

Part 4: Metadata and Ownership

Storing Metadata

# Store metadata for a schema
metadata = {
    "title": "User Schema",
    "description": "Defines the structure of user records",
    "domain": "customer",
    "data_classification": "pii",
    "retention_period": "7 years",
    "business_owners": ["product-team"],
    "technical_owners": ["data-engineering"],
    "steward": "alice@example.com",
    "tags": ["user", "customer", "pii"]
}

store.store_metadata("user", "1.0.0", metadata)

Retrieving Metadata

# Get metadata (by contract name and version)
metadata = store.get_metadata("user", "1.0.0")
print(f"Title: {metadata.get('title')}")
print(f"Owner: {metadata.get('business_owners')}")
print(f"Tags: {metadata.get('tags')}")

To find contracts by tag or owner, use list_contracts() and filter the list in your application, or query the store's backend directly.

Part 5: Using with Validator

Creating Validators from Store

from pycharter import Validator, SQLiteContractStore

store = SQLiteContractStore("metadata.db")
store.connect()

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

# Validate data
result = validator.validate({
    "id": "123",      # Will be coerced
    "name": "Alice",
    "email": "ALICE@EXAMPLE.COM",  # Will be lowercased
    "age": "30"       # Will be coerced
})

if result.is_valid:
    print(f"Valid: {result.data}")

Building Contracts from Store

from pycharter import build_contract_from_store

# Build a consolidated contract
# Note: Contract has RAW schema + separate rules (not merged)
contract = build_contract_from_store(store, "user", "1.0.0")
# contract is a DataContract (use .json_schema, .coercion_rules, etc.)
print(f"Schema (raw): {contract.json_schema}")
print(f"Coercion rules (separate): {contract.coercion_rules}")
print(f"Validation rules (separate): {contract.validation_rules}")

Part 6: Schema Evolution

Checking Compatibility

from pycharter.schema_evolution import check_compatibility, CompatibilityMode

# Check if new schema is backward compatible
old_schema = store.get_schema("user", "1.0.0")
new_schema = store.get_schema("user", "2.0.0")

result = check_compatibility(
    old_schema=old_schema,
    new_schema=new_schema,
    mode=CompatibilityMode.BACKWARD
)

print(f"Compatible: {result.is_compatible}")
if not result.is_compatible:
    for issue in result.issues:
        print(f"  - {issue}")

Computing Schema Diff

from pycharter.schema_evolution import compute_diff

diff = compute_diff(old_schema, new_schema)

print("Added fields:")
for field in diff.added_fields:
    print(f"  + {field}")

print("Removed fields:")
for field in diff.removed_fields:
    print(f"  - {field}")

print("Modified fields:")
for field, changes in diff.modified_fields.items():
    print(f"  ~ {field}: {changes}")

Migration Strategies

# Check compatibility before storing new version
def store_schema_safe(store, name, schema, version, previous_version=None):
    """Store schema only if backward compatible with previous_version."""

    if previous_version:
        current = store.get_schema(name, previous_version)

        # Check compatibility
        result = check_compatibility(
            old_schema=current,
            new_schema=schema,
            mode=CompatibilityMode.BACKWARD
        )

        if not result.is_compatible:
            raise ValueError(
                f"Schema not backward compatible: {result.issues}"
            )

    return store.store_schema(name, version, schema)

Part 7: Database Initialization

CLI Commands

# Initialize database (create tables)
pycharter db init

# Run migrations
pycharter db migrate

# Check migration status
pycharter db status

# Rollback last migration
pycharter db rollback

Programmatic Initialization

from pycharter.db import init_database, run_migrations

# Initialize database
init_database("postgresql://user:pass@localhost/pycharter")

# Run migrations
run_migrations("postgresql://user:pass@localhost/pycharter")

Part 8: Best Practices

1. Use Semantic Versioning

# Follow semver: MAJOR.MINOR.PATCH
# MAJOR: Breaking changes
# MINOR: Backward-compatible additions
# PATCH: Backward-compatible fixes

store.store_schema("user", "1.0.0", schema)  # Initial
store.store_schema("user", "1.1.0", schema)   # Added optional field
store.store_schema("user", "2.0.0", schema)   # Breaking change

2. Document Ownership

# Always include ownership metadata
metadata = {
    "business_owners": ["product-team"],
    "technical_owners": ["data-engineering"],
    "steward": "alice@example.com",
    "slack_channel": "#data-contracts",
    "oncall_rotation": "data-team"
}

3. Tag for Discovery

# Use consistent tags
metadata = {
    "tags": [
        "customer",           # Domain
        "pii",               # Data classification
        "production",        # Environment
        "v2"                 # Major version
    ]
}

4. Track Lineage

# Include lineage in metadata
metadata = {
    "source_systems": ["crm", "billing"],
    "downstream_consumers": ["analytics", "marketing"],
    "data_product": "customer-360"
}

5. Automate Sync

# Sync schemas from contract files on deploy
from pathlib import Path

def sync_contracts(store, contracts_dir):
    """Sync all contract files to the contract store."""

    for contract_file in Path(contracts_dir).glob("**/*.yaml"):
        with open(contract_file) as f:
            contract = yaml.safe_load(f)

        name = contract_file.stem
        version = contract["json_schema"]["version"]

        store.store_schema(name, version, contract["json_schema"])

        if contract.get("coercion_rules"):
            store.store_coercion_rules(name, version, contract["coercion_rules"])

        if contract.get("validation_rules"):
            store.store_validation_rules(name, version, contract["validation_rules"])

        if contract.get("metadata"):
            store.store_metadata(name, version, contract["metadata"])

        print(f"Synced {name} v{version}")

Exercises

  1. Setup: Create a SQLite store and store a schema with coercion/validation rules.

  2. Versioning: Store multiple versions of a schema and retrieve specific versions.

  3. Metadata: Add ownership and governance metadata, then search by tags.

  4. Evolution: Check compatibility between two schema versions and compute the diff.

  5. Integration: Create a validator from the store and validate a batch of records.

Next Steps