Contract Store¶

The contract store provides centralized schema registry functionality. Contract identity is namespace, 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_json_schema("user", "1.0.0", schema)
store.store_metadata("user", "1.0.0", metadata)
schema = store.get_json_schema("user", "1.0.0")
contracts = store.list_contracts()  # [{"name": "user", "version": "1.0.0"}, ...]
store.disconnect()

Namespace and latest-version resolution¶

Pass namespace="market" to select a namespace explicitly. Unqualified latest_version(name) and get(name) raise AmbiguousContractError when the name appears in multiple namespaces, even when only one namespace has an active version. Empty and whitespace namespace arguments remain unqualified. An explicit version may resolve unqualified when that (name, version) exists in exactly one namespace; specify the namespace if that pair is ambiguous.

Within one resolved namespace, latest_version() prefers the newest version whose status is neither deprecated nor retired. If all versions have those statuses, it returns the absolute latest version for compatibility. Use an explicit version when a reference must remain pinned.

Version precedence and selection ties¶

Latest selection uses SemVer precedence: numeric dotted prerelease identifiers compare numerically (rc.10 > rc.2), numeric identifiers precede nonnumeric ones, and a release follows its prereleases. Build metadata is accepted but does not affect precedence. When two stored spellings have equal precedence, the lexicographically greatest original version string wins selection, independent of insertion/backend order. For example, 1.0.0+build.z is selected over 1.0.0+build.a, but is not a semantically newer version. Exact pins retain their original identity.

compare_versions() and is_version_higher() compare precedence only. get_latest_version() and version_sort_key() apply the documented textual selection tie-break. Proposal admission uses precedence, so changing only build metadata or a legacy spelling cannot satisfy a required version increase.

The compatibility helpers remain tolerant: 1, 1.2, leading v/V, and previously accepted legacy spellings still parse. They do not replace native artifact SemVer validation. parse_version() retains its four-component return value and omits build metadata. The public sort key remains a six-tuple of five integers and a string; its final string is now an opaque sortable encoding rather than a prerelease label. Do not persist or interpret that string as an artifact version. Unparseable strings still sort before parseable versions and lexically among themselves. Comparison still raises on invalid versions; get_latest_version() retains its empty/singleton behavior and raises on invalid entries when comparing multiple versions.

The same helper serves pipeline stores, semantic scheme/document lookups, agent preparation, and native DAG references. Their lifecycle policies are unchanged: contract latest prefers non-deprecated/non-retired versions, preparation requires active versions, and a DAG's unpinned reference selects the absolute latest known version. Namespace ambiguity remains an error before contract lifecycle or version filtering. Semantic-store acceptance checks ordering of existing versioned rows; semantic mutation authority remains a separate backlog item.

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: VersionedArtifactStore

Client for storing and retrieving versioned data contracts.

All operations are keyed by (contract_name, contract_version). Implementations extend this for specific backends (PostgreSQL, SQLite, MongoDB, Redis, InMemory).

The shared :class:~pycharter.stores.base.VersionedArtifactStore surface (save / get / list / list_versions / set_status / update / delete) is the single public write path. save writes the whole contract atomically; update applies a sectional patch. The contract document carries::

{json_schema, coercion_rules, validation_rules, metadata,
 field_bindings, data_quality, sla_default_element, sla_properties}

json_schema is required on :meth:save; the rest are optional.

The per-artifact getters (get_json_schema, get_metadata, get_coercion_rules, get_validation_rules, get_field_bindings, get_data_quality, get_sla, get_complete_json_schema, get_contract) remain public; the per-artifact writers are internal to each backend's atomic write path.

NS-2/NS-3 — the full identity is (namespace, name, version) where namespace=None (or "") means the global namespace. Identity-based reads and writes accept a keyword-only namespace:

namespace given: exact (namespace, name, version) match.
namespace omitted: match by (name, version); when the name exists in exactly one namespace it resolves as before, and when several namespaces hold it the lookup raises :class:~pycharter.stores.errors.AmbiguousContractError listing the candidates.

On :meth:save the namespace persists from the explicit keyword, else from the document's metadata.namespace, else the global namespace. The stored metadata payload keeps its namespace key (the document is the source of truth; the stored identity mirrors it).

Parameters:

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

connect `abstractmethod` ¶

connect() -> None

Establish the backend connection.

disconnect ¶

disconnect() -> None

Close the backend connection.

store_json_schema `abstractmethod` ¶

store_json_schema(
    contract_name: str,
    contract_version: str,
    schema: dict[str, Any],
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Contract namespace; a newly created contract is placed in it (`None` resolves by name, creating in the global namespace when absent).	`None`

Returns:

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

get_json_schema `abstractmethod` ¶

get_json_schema(
    contract_name: str,
    contract_version: str,
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

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],
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

Returns:

Type	Description
`str`	Coercion rules ID or identifier.

get_coercion_rules `abstractmethod` ¶

get_coercion_rules(
    contract_name: str,
    contract_version: str,
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

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],
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

Returns:

Type	Description
`str`	Validation rules ID or identifier.

get_validation_rules `abstractmethod` ¶

get_validation_rules(
    contract_name: str,
    contract_version: str,
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

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],
    *,
    namespace: str | None = None,
) -> str

Store metadata for the given data contract.

NS-2: the document is the source of truth — a payload carrying namespace stamps a contract still in the global namespace.

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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

Returns:

Type	Description
`str`	Metadata record ID or identifier.

get_metadata `abstractmethod` ¶

get_metadata(
    contract_name: str,
    contract_version: str,
    *,
    namespace: str | None = None,
) -> 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

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,
    *,
    namespace: str | None = None,
) -> 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_json_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`
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

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_bindings ¶

get_field_bindings(
    contract_name: str,
    contract_version: str,
    *,
    namespace: str | None = None,
) -> dict[str, Any] | None

Retrieve field bindings 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
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

Returns:

Type	Description
`dict[str, Any] \| None`	Field-bindings artifact dictionary or None if not found.

store_field_bindings ¶

store_field_bindings(
    contract_name: str,
    contract_version: str,
    field_bindings: dict[str, Any],
    *,
    namespace: str | None = None,
) -> str

Store a field bindings 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_bindings`	`dict[str, Any]`	Field bindings artifact payload.	required
`namespace`	`str \| None`	Exact namespace, or `None` to resolve by name.	`None`

Returns:

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

SQLiteContractStore¶

from pycharter import SQLiteContractStore

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

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

Existing SQLite databases: run pycharter db upgrade

Earlier releases declared UUID columns as UUID on SQLite, a type SQLite treats as numeric. Roughly one UUID in a million has all-digit hex, and SQLite stored those as floating-point numbers: the digits were lost and the row could no longer be read. Databases created now are not affected.

Run pycharter db upgrade <database-url> once on an existing SQLite database. The migration retypes the UUID columns to CHAR(32) without rewriting data, and reports any value that was already corrupted (it cannot be recovered). This also works for a database created by connect(auto_initialize=True), which it brings under migrations, adding the rest of pycharter's tables. PostgreSQL uses a native uuid type and is unaffected.

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_json_schema("user", "1.0.0", schema_v1)
store.store_json_schema("user", "2.0.0", schema_v2)

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

# List contracts
contracts = store.list_contracts()

A version is immutable where it decides a verdict: changing its schema, coercion or validation rules, field bindings, data_quality.rules or strict_types raises ArtifactContentConflictError, so publish a new version instead. See Artifact format versioning.

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

Field bindings artifact¶

Stores that support semantic bindings (e.g. PostgresContractStore, InMemoryContractStore) persist the versioned field bindings artifact per contract.

# Store field bindings for a contract
store.store_field_bindings(
    "user",
    "1.0.0",
    {
        "version": "1.0.0",
        "domain_schema": "trading-domain:1.0.0",
        "field_bindings": {
            "$.email": {
                "primary": {
                    "concept": {"concept": "user_email"},
                    "concept_scheme": "trading-domain:1.0.0",
                }
            }
        },
    },
)

# Retrieve
field_bindings = store.get_field_bindings("user", "1.0.0")

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

Breaking change: legacy keys field_mapping, ontology (contract artifact), and store methods get_field_mapping / store_field_mapping were removed. Run pycharter db upgrade to rename field_mappings → field_bindings in PostgreSQL.

Contract Store¶

Overview¶

Namespace and latest-version resolution¶

Version precedence and selection ties¶

Store Implementations¶

ContractStoreClient¶

ContractStoreClient ¶

connect abstractmethod ¶

disconnect ¶

store_json_schema abstractmethod ¶

get_json_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_bindings ¶

store_field_bindings ¶

SQLiteContractStore¶

PostgresContractStore¶

MongoDBContractStore¶

RedisContractStore¶

InMemoryContractStore¶

Examples¶

Schema Versioning¶

Storing Rules¶

Metadata¶

Field bindings artifact¶

See Also¶

connect `abstractmethod` ¶

store_json_schema `abstractmethod` ¶

get_json_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` ¶