Contract Store¶

The contract store provides centralized schema registry functionality. All operations are keyed by contract name and contract version.

Overview¶

from pycharter import PostgresContractStore

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

# Store and retrieve by contract name and version
store.store_schema("user", "1.0.0", schema)
store.store_metadata("user", "1.0.0", metadata)
schema = store.get_schema("user", "1.0.0")
contracts = store.list_contracts()  # [{"name": "user", "version": "1.0.0"}, ...]
store.disconnect()

Store Implementations¶

Class	Backend	Use Case
`InMemoryContractStore`	Memory	Testing
`SQLiteContractStore`	SQLite	Development
`PostgresContractStore`	PostgreSQL	Production
`MongoDBContractStore`	MongoDB	Document-oriented
`RedisContractStore`	Redis	Caching

ContractStoreClient¶

Base interface for all stores:

ContractStoreClient ¶

ContractStoreClient(connection_string: str | None = None)

Bases: ABC

Client for storing and retrieving data contracts. All operations are keyed by data contract (contract_name, contract_version). Implementations extend this for specific backends (PostgreSQL, SQLite, MongoDB, etc.).

Parameters:

Name	Type	Description	Default
`connection_string`	`str \| None`	Database connection string (format depends on implementation)	`None`

connect `abstractmethod` ¶

connect() -> None

Establish database connection. Subclasses must implement.

disconnect ¶

disconnect() -> None

Close database connection.

store_schema `abstractmethod` ¶

store_schema(contract_name: str, contract_version: str, schema: dict[str, Any]) -> str

Store a JSON Schema for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required
`schema`	`dict[str, Any]`	JSON Schema dictionary (may contain "version" for schema artifact version).	required

Returns:

Type	Description
`str`	Contract ID or identifier (implementation-defined).

get_schema `abstractmethod` ¶

get_schema(contract_name: str, contract_version: str) -> dict[str, Any]

Retrieve the schema for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required

Returns:

Type	Description
`dict[str, Any]`	Schema dictionary with version included, or None if not found.

list_contracts `abstractmethod` ¶

list_contracts() -> list[dict[str, Any]]

List all stored data contracts.

Returns:

Type	Description
`list[dict[str, Any]]`	List of dicts with at least "name" and "version"; may include id, schema_id, etc.

store_coercion_rules `abstractmethod` ¶

store_coercion_rules(contract_name: str, contract_version: str, coercion_rules: dict[str, Any]) -> str

Store coercion rules for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required
`coercion_rules`	`dict[str, Any]`	Dict mapping field names to coercion function names.	required

Returns:

Type	Description
`str`	Coercion rules ID or identifier.

get_coercion_rules `abstractmethod` ¶

get_coercion_rules(contract_name: str, contract_version: str) -> dict[str, Any]

Retrieve coercion rules for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required

Returns:

Type	Description
`dict[str, Any]`	Coercion rules dictionary or None if not found.

store_validation_rules `abstractmethod` ¶

store_validation_rules(contract_name: str, contract_version: str, validation_rules: dict[str, Any]) -> str

Store validation rules for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required
`validation_rules`	`dict[str, Any]`	Dict mapping field names to validation configs.	required

Returns:

Type	Description
`str`	Validation rules ID or identifier.

get_validation_rules `abstractmethod` ¶

get_validation_rules(contract_name: str, contract_version: str) -> dict[str, Any]

Retrieve validation rules for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required

Returns:

Type	Description
`dict[str, Any]`	Validation rules dictionary or None if not found.

store_metadata `abstractmethod` ¶

store_metadata(contract_name: str, contract_version: str, metadata: dict[str, Any]) -> str

Store metadata for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required
`metadata`	`dict[str, Any]`	Metadata dictionary.	required

Returns:

Type	Description
`str`	Metadata record ID or identifier.

get_metadata `abstractmethod` ¶

get_metadata(contract_name: str, contract_version: str) -> dict[str, Any]

Retrieve metadata for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required

Returns:

Type	Description
`dict[str, Any]`	Metadata dictionary or None if not found.

get_contract ¶

get_contract(contract_name: str, contract_version: str, include_metadata: bool = True, include_ownership: bool = True, include_governance: bool = True) -> dict[str, Any]

Return the full data contract (schema + coercion_rules + validation_rules + metadata) for the given contract name and version.

Default implementation assembles from get_schema, get_coercion_rules, get_validation_rules, get_metadata. Subclasses may override for efficiency.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required
`include_metadata`	`bool`	Whether to include metadata in the contract.	`True`
`include_ownership`	`bool`	Whether to include ownership from metadata.	`True`
`include_governance`	`bool`	Whether to include governance_rules from metadata.	`True`

Returns:

Type	Description
`dict[str, Any]`	Consolidated contract dict (schema, coercion_rules, validation_rules,
`dict[str, Any]`	metadata, versions) or None if contract/schema not found.

get_field_mapping ¶

get_field_mapping(contract_name: str, contract_version: str) -> dict[str, Any] | None

Retrieve field mapping for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required

Returns:

Type	Description
`dict[str, Any] \| None`	FieldMapping dictionary or None if not found.

store_field_mapping ¶

store_field_mapping(contract_name: str, contract_version: str, field_mapping: dict[str, Any]) -> str

Store a field mapping artifact for the given data contract.

Parameters:

Name	Type	Description	Default
`contract_name`	`str`	Data contract name.	required
`contract_version`	`str`	Data contract version.	required
`field_mapping`	`dict[str, Any]`	FieldMapping dict (version, field_bindings, etc.).	required

Returns:

Type	Description
`str`	Identifier for the stored field mapping.

SQLiteContractStore¶

from pycharter import SQLiteContractStore

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

# Store and retrieve by contract name and version
store.store_schema("user", "1.0.0", {
    "type": "object",
    "properties": {"name": {"type": "string"}}
})
schema = store.get_schema("user", "1.0.0")

PostgresContractStore¶

from pycharter import PostgresContractStore

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

MongoDBContractStore¶

from pycharter import MongoDBContractStore

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

RedisContractStore¶

from pycharter import RedisContractStore

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

InMemoryContractStore¶

from pycharter import InMemoryContractStore

store = InMemoryContractStore()
store.connect()
# Data is lost when program exits

Examples¶

Schema Versioning¶

# Store versions
store.store_schema("user", "1.0.0", schema_v1)
store.store_schema("user", "2.0.0", schema_v2)

# Get specific version
schema = store.get_schema("user", "1.0.0")

# List contracts
contracts = store.list_contracts()

Storing Rules¶

# Coercion rules
store.store_coercion_rules("user", "1.0.0", {
    "version": "1.0.0",
    "rules": {"age": "coerce_to_integer"}
})

# Validation rules
store.store_validation_rules("user", "1.0.0", {
    "version": "1.0.0",
    "rules": {"age": {"is_positive": {}}}
})

Metadata¶

# Store metadata
store.store_metadata("user", "1.0.0", {
    "title": "User Schema",
    "owner": "data-team",
    "tags": ["user", "pii"]
})

# Retrieve
metadata = store.get_metadata("user", "1.0.0")

Ontology (ontology-enabled stores)¶

Stores that support ontology (e.g. PostgresContractStore, InMemoryContractStore) can store and retrieve ontology annotations per contract. PostgresContractStore requires the semantic/ontology tables (e.g. contract_versions) to exist; run pycharter db upgrade if ontology storage fails.

# Store field mapping (ontology) for a contract
store.store_field_mapping("user", "1.0.0", {
    "version": "1.0.0",
    "fields": {
        "email": {"concept": "user_email", "definition": "Primary email address"}
    }
})

# Retrieve
field_mapping = store.get_field_mapping("user", "1.0.0")

When you call store.get_contract(name, version), the returned contract dict includes field_mapping when available.

Contract Store¶

Overview¶

Store Implementations¶

ContractStoreClient¶

ContractStoreClient ¶

connect abstractmethod ¶

disconnect ¶

store_schema abstractmethod ¶

get_schema abstractmethod ¶

list_contracts abstractmethod ¶

store_coercion_rules abstractmethod ¶

get_coercion_rules abstractmethod ¶

store_validation_rules abstractmethod ¶

get_validation_rules abstractmethod ¶

store_metadata abstractmethod ¶

get_metadata abstractmethod ¶

get_contract ¶

get_field_mapping ¶

store_field_mapping ¶

SQLiteContractStore¶

PostgresContractStore¶

MongoDBContractStore¶

RedisContractStore¶

InMemoryContractStore¶

Examples¶

Schema Versioning¶

Storing Rules¶

Metadata¶

Ontology (ontology-enabled stores)¶

See Also¶

connect `abstractmethod` ¶

store_schema `abstractmethod` ¶

get_schema `abstractmethod` ¶

list_contracts `abstractmethod` ¶

store_coercion_rules `abstractmethod` ¶

get_coercion_rules `abstractmethod` ¶

store_validation_rules `abstractmethod` ¶

get_validation_rules `abstractmethod` ¶

store_metadata `abstractmethod` ¶

get_metadata `abstractmethod` ¶