Contract Builder

Build consolidated contracts from separate artifacts. The builder returns a DataContract (first-class Pydantic type) used by the validator, quality checks, and ETL.

Overview

from pycharter import (
    build_contract,
    build_contract_from_store,
    ContractArtifacts,
    DataContract,
)

# From artifacts → DataContract
artifacts = ContractArtifacts(
    json_schema=schema, coercion_rules=coercion, validation_rules=validation
)
contract = build_contract(artifacts)  # returns DataContract

# From store (by contract name and version) → DataContract
contract = build_contract_from_store(store, "user", "1.0.0")

Contract Structure

Schema is RAW

The contract (DataContract or dict) contains raw schema + separate rules. Rules are NOT merged into the schema. The Validator class handles merging internally during validation.

The returned DataContract (or .to_dict()) has this structure:

{
    "json_schema": {...},  # RAW JSON Schema (no coercion/validation merged)
    "coercion_rules": {...},  # Separate coercion rules (always present, may be {})
    "validation_rules": {...},  # Separate validation rules (always present, may be {})
    "metadata": {...},  # Optional metadata
    "ownership": {...},  # Optional ownership
    "governance_rules": {...},  # Optional governance
    "field_bindings": {...},  # Optional field bindings artifact
    "versions": {...},  # Version tracking
}

To get a merged schema (for display or legacy code), use:

from pycharter import get_merged_schema_from_contract

merged_schema = get_merged_schema_from_contract(contract)
# merged_schema["properties"]["age"]["coercion"] now exists

API Reference

build_contract

build_contract(
    artifacts: ContractArtifacts,
    include_metadata: bool = True,
    include_ownership: bool = True,
    include_governance: bool = True,
) -> DataContract

Build a consolidated data contract from separate artifacts.

This function combines schema, coercion rules, validation rules, and metadata into a single consolidated contract. It tracks versions of all components.

IMPORTANT: The schema in the returned contract is RAW (not merged with rules). The Validator class handles merging rules into the schema internally during validation. To get a merged schema explicitly, use merge_rules_into_schema() from pycharter.runtime_validator.utils.

Parameters:

Name	Type	Description	Default
artifacts	ContractArtifacts	ContractArtifacts object containing all separate artifacts	required
include_metadata	bool	Whether to include metadata in the contract	True
include_ownership	bool	Whether to include ownership information	True
include_governance	bool	Whether to include governance rules	True

Returns:

Name	Type	Description
	DataContract	DataContract with:
	DataContract	json_schema: RAW JSON Schema (no coercion/validation merged; must have version)
	DataContract	coercion_rules: Coercion rules dictionary (always present, may be empty)
	DataContract	validation_rules: Validation rules dictionary (always present, may be empty)
	DataContract	metadata: Metadata containing ownership and governance_rules (if include_metadata=True)
	DataContract	field_bindings: Semantic field bindings (if present in artifacts)
	DataContract	versions: Dictionary tracking versions of all components
Note	DataContract	ownership and governance_rules are nested inside metadata, not at top level.

Raises:

Type	Description
ValueError	If schema does not have a version field

Example

artifacts = ContractArtifacts( ... json_schema={"type": "object", "version": "1.0.0", "properties": {"age": {"type": "integer"}}}, ... coercion_rules={"version": "1.0.0", "rules": {"age": "coerce_to_integer"}}, ... validation_rules={"version": "1.0.0", "rules": {}}, ... metadata={"version": "1.0.0", "description": "User contract"}, ... ownership={"owner": "data-team"}, ... governance_rules={} ... ) contract = build_contract(artifacts) sorted((contract.versions or {}).keys()) ['coercion_rules', 'json_schema', 'metadata', 'validation_rules'] "coercion" not in (contract.json_schema.get("properties") or {}).get("age", {}) True

Source code in src/pycharter/contract_builder/builder.py

def build_contract(
    artifacts: ContractArtifacts,
    include_metadata: bool = True,
    include_ownership: bool = True,
    include_governance: bool = True,
) -> DataContract:
    """
    Build a consolidated data contract from separate artifacts.

    This function combines schema, coercion rules, validation rules, and metadata
    into a single consolidated contract. It tracks versions of all components.

    IMPORTANT: The schema in the returned contract is RAW (not merged with rules).
    The Validator class handles merging rules into the schema internally during
    validation. To get a merged schema explicitly, use merge_rules_into_schema()
    from pycharter.runtime_validator.utils.

    Args:
        artifacts: ContractArtifacts object containing all separate artifacts
        include_metadata: Whether to include metadata in the contract
        include_ownership: Whether to include ownership information
        include_governance: Whether to include governance rules

    Returns:
        DataContract with:
        - json_schema: RAW JSON Schema (no coercion/validation merged; must have version)
        - coercion_rules: Coercion rules dictionary (always present, may be empty)
        - validation_rules: Validation rules dictionary (always present, may be empty)
        - metadata: Metadata containing ownership and governance_rules (if include_metadata=True)
        - field_bindings: Semantic field bindings (if present in artifacts)
        - versions: Dictionary tracking versions of all components

        Note: ownership and governance_rules are nested inside metadata, not at top level.

    Raises:
        ValueError: If schema does not have a version field

    Example:
        >>> artifacts = ContractArtifacts(
        ...     json_schema={"type": "object", "version": "1.0.0", "properties": {"age": {"type": "integer"}}},
        ...     coercion_rules={"version": "1.0.0", "rules": {"age": "coerce_to_integer"}},
        ...     validation_rules={"version": "1.0.0", "rules": {}},
        ...     metadata={"version": "1.0.0", "description": "User contract"},
        ...     ownership={"owner": "data-team"},
        ...     governance_rules={}
        ... )
        >>> contract = build_contract(artifacts)
        >>> sorted((contract.versions or {}).keys())
        ['coercion_rules', 'json_schema', 'metadata', 'validation_rules']
        >>> "coercion" not in (contract.json_schema.get("properties") or {}).get("age", {})
        True
    """
    if not artifacts.json_schema:
        raise ValueError("json_schema is required to build a contract")

    # Validate schema has version
    if "version" not in artifacts.json_schema:
        raise ValueError(
            "Schema must have a 'version' field. All schemas must be versioned. "
            "Please ensure the schema dictionary contains 'version': '<version_string>'."
        )

    # Deep copy schema to avoid modifying original - NO MERGING
    # The Validator class handles merging internally during validation
    raw_schema = copy.deepcopy(artifacts.json_schema)

    # Extract rules (but do NOT merge into schema)
    coercion_rules = _extract_rules(artifacts.coercion_rules) or {}
    validation_rules = _extract_rules(artifacts.validation_rules) or {}

    # Build contract with raw schema and separate rules
    contract: dict[str, Any] = {
        "json_schema": raw_schema,
        "coercion_rules": copy.deepcopy(coercion_rules) if coercion_rules else {},
        "validation_rules": copy.deepcopy(validation_rules) if validation_rules else {},
    }

    # Track versions
    versions: dict[str, str | None] = {}

    # Extract version from schema
    schema_version = _extract_version(artifacts.json_schema)
    if schema_version:
        versions["json_schema"] = schema_version

    # Extract version from coercion rules
    if artifacts.coercion_rules:
        coercion_version = _extract_version(artifacts.coercion_rules)
        if coercion_version:
            versions["coercion_rules"] = coercion_version

    # Extract version from validation rules
    if artifacts.validation_rules:
        validation_version = _extract_version(artifacts.validation_rules)
        if validation_version:
            versions["validation_rules"] = validation_version

    # Extract version from metadata
    if artifacts.metadata:
        metadata_version = _extract_version(artifacts.metadata)
        if metadata_version:
            versions["metadata"] = metadata_version

    # Extract version from field_bindings
    if artifacts.field_bindings:
        mapping_version = _extract_version(artifacts.field_bindings)
        if mapping_version:
            versions["field_bindings"] = mapping_version

    # Add metadata if requested
    if include_metadata and artifacts.metadata:
        # Create metadata dict without version (version is tracked separately)
        metadata_dict = copy.deepcopy(artifacts.metadata)
        metadata_dict.pop("version", None)  # Remove version, it's in versions dict

        # Add ownership inside metadata if requested
        if include_ownership and artifacts.ownership:
            metadata_dict["ownership"] = copy.deepcopy(artifacts.ownership)

        # Add governance rules inside metadata if requested
        if include_governance and artifacts.governance_rules:
            metadata_dict["governance_rules"] = copy.deepcopy(
                artifacts.governance_rules
            )

        contract["metadata"] = metadata_dict

    # Also expose ownership and governance_rules at top level for easier access
    if include_ownership and artifacts.ownership:
        contract["ownership"] = copy.deepcopy(artifacts.ownership)

    if include_governance and artifacts.governance_rules:
        contract["governance_rules"] = copy.deepcopy(artifacts.governance_rules)

    # Add field_bindings if present
    if artifacts.field_bindings:
        contract["field_bindings"] = copy.deepcopy(artifacts.field_bindings)

    # Add versions tracking
    if versions:
        contract["versions"] = versions

    return DataContract.model_validate(contract)

build_contract_from_store

build_contract_from_store(
    store: ContractStoreClient,
    contract_name: str,
    contract_version: str,
    include_metadata: bool = True,
    include_ownership: bool = True,
    include_governance: bool = True,
    *,
    namespace: str | None = None,
) -> DataContract

Build a consolidated data contract from the contract store by contract name and version.

Uses store.get_contract(contract_name, contract_version, ...) to load the full contract (schema + coercion_rules + validation_rules + metadata + field_bindings).

Parameters:

Name	Type	Description	Default
store	ContractStoreClient	ContractStoreClient instance (contract-centric)	required
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 information	True
include_governance	bool	Whether to include governance rules	True
namespace	str | None	Exact contract namespace, or None to resolve by name (NS-3 name-only semantics)	None

Returns:

Type	Description
DataContract	DataContract ready for runtime validation

Raises:

Type	Description
ValueError	If the contract (or its schema) is not found
AmbiguousContractError	If namespace is omitted and the name exists in several namespaces

Source code in src/pycharter/contract_builder/builder.py

def build_contract_from_store(
    store: ContractStoreClient,
    contract_name: str,
    contract_version: str,
    include_metadata: bool = True,
    include_ownership: bool = True,
    include_governance: bool = True,
    *,
    namespace: str | None = None,
) -> DataContract:
    """
    Build a consolidated data contract from the contract store by contract name and version.

    Uses store.get_contract(contract_name, contract_version, ...) to load the full
    contract (schema + coercion_rules + validation_rules + metadata + field_bindings).

    Args:
        store: ContractStoreClient instance (contract-centric)
        contract_name: Data contract name
        contract_version: Data contract version
        include_metadata: Whether to include metadata in the contract
        include_ownership: Whether to include ownership information
        include_governance: Whether to include governance rules
        namespace: Exact contract namespace, or ``None`` to resolve by
            name (NS-3 name-only semantics)

    Returns:
        DataContract ready for runtime validation

    Raises:
        ValueError: If the contract (or its schema) is not found
        AmbiguousContractError: If ``namespace`` is omitted and the name
            exists in several namespaces
    """
    envelope = store.get(contract_name, contract_version, namespace=namespace)
    if not envelope or not envelope.get("document"):
        raise ValueError(
            f"Contract not found: {contract_name!r} @ {contract_version!r}. "
            "Ensure the schema is stored for this contract."
        )
    contract = copy.deepcopy(envelope["document"])
    # The envelope document always carries the full contract; honor the
    # include flags here so the build surface matches the prior
    # ``get_contract(..., include_*=...)`` behavior.
    if not include_metadata:
        contract.pop("metadata", None)
    if not include_ownership:
        contract.pop("ownership", None)
        if isinstance(contract.get("metadata"), dict):
            contract["metadata"].pop("ownership", None)
    if not include_governance:
        contract.pop("governance_rules", None)
        if isinstance(contract.get("metadata"), dict):
            contract["metadata"].pop("governance_rules", None)
    if include_metadata:
        # Component metadata may omit its own version. The envelope holds the
        # exact contract identity; component versions remain in ``versions``.
        metadata = contract.get("metadata")
        if metadata is None:
            metadata = {}
            contract["metadata"] = metadata
        if isinstance(metadata, dict):
            metadata["name"] = envelope["name"]
            metadata["version"] = envelope["version"]
            if envelope.get("namespace") is not None:
                metadata["namespace"] = envelope["namespace"]
    if not contract.get("json_schema"):
        raise ValueError(
            f"Contract {contract_name!r} @ {contract_version!r} has no json_schema."
        )
    if "version" not in contract["json_schema"]:
        raise ValueError(
            f"Contract {contract_name!r} @ {contract_version!r} schema does not have a version field."
        )
    return DataContract.model_validate(contract)

ContractArtifacts

Input for building contracts:

Attribute	Type	Description
json_schema	Dict	JSON Schema (will be stored as-is, not modified)
coercion_rules	Dict	Coercion rules (stored separately)
validation_rules	Dict	Validation rules (stored separately)
metadata	Dict	Metadata
ownership	Dict	Ownership
governance_rules	Dict	Governance
field_bindings	Dict	Field bindings artifact (version, domain_schema, field_bindings map)

Examples

from pycharter import (
    build_contract,
    build_contract_from_store,
    ContractArtifacts,
    Validator,
)

# Build from artifacts
artifacts = ContractArtifacts(
    json_schema={
        "type": "object",
        "version": "1.0.0",
        "properties": {"age": {"type": "integer"}},
    },
    coercion_rules={"rules": {"age": "coerce_to_integer"}},
    validation_rules={"rules": {"age": {"is_positive": {"threshold": 0}}}},
)
contract = build_contract(artifacts)

# DataContract is a Pydantic model — use attributes or .to_dict()
print(
    contract.json_schema["properties"]["age"]
)  # {"type": "integer"} — no coercion merged
print(contract.coercion_rules)  # {"rules": {"age": "coerce_to_integer"}}

# Use Validator (accepts DataContract or dict)
validator = Validator(contract)
result = validator.validate({"age": "25"})  # Coercion applied automatically

# Build from store (by contract name and version)
contract = build_contract_from_store(store, "user", "1.0.0")

DataContract

The first-class runtime contract type. Builders return it; parsed contracts expose .to_data_contract(). Validator, QualityCheck, and ETL accept it or its runtime dict form via .to_dict().

DataContract

Bases: BaseModel

Consolidated data contract: JSON Schema, rules, metadata, and field bindings.

This is the single source of truth used by Validator, ETL validation, and quality checks. json_schema is raw (not merged with rules); the Validator merges rules internally during validation.

json_schema class-attribute instance-attribute

json_schema: dict[str, Any] = Field(
    ...,
    description="JSON Schema definition (raw, must have version)",
)

coercion_rules class-attribute instance-attribute

coercion_rules: dict[str, Any] = Field(
    default_factory=dict, description="Coercion rules"
)

validation_rules class-attribute instance-attribute

validation_rules: dict[str, Any] = Field(
    default_factory=dict, description="Validation rules"
)

metadata class-attribute instance-attribute

metadata: dict[str, Any] | None = Field(
    default=None,
    description="Metadata (ownership, governance_rules)",
)

field_bindings class-attribute instance-attribute

field_bindings: dict[str, Any] | None = Field(
    default=None,
    description="Semantic field bindings artifact",
)

to_dict

to_dict() -> dict[str, Any]

Return contract as a plain dict for serialization or legacy callers.

Source code in src/pycharter/contract_builder/types.py

def to_dict(self) -> dict[str, Any]:
    """Return contract as a plain dict for serialization or legacy callers."""
    return self.model_dump()

Serialize a canonical contract

Use the canonical dump functions to exchange a built contract as a document:

from pycharter.semantic.contracts import (
    dump_contract,
    dump_contract_to_dict,
    dump_contract_to_string,
    load_contract,
)

contract = build_contract_from_store(store, "user", "1.0.0")
document = dump_contract_to_dict(contract)
yaml_text = dump_contract_to_string(contract)
dump_contract(contract, "contract.yaml")
parsed = load_contract("contract.yaml")

These functions accept both DataContract and parsed ContractMetadata. For a built model, they omit top-level null fields and place ownership and governance convenience values inside metadata. Existing nested metadata keys take precedence, including explicit null or empty values. When the original model metadata exists but lacks version, its known component-ledger version is retained in the document. Schema versions remain independent. Nested payload nulls, false values, empty collections, and explicit format markers are preserved.

DataContract.to_dict() and model_dump() retain their runtime representation, including top-level convenience fields. The canonical projection produces independent dictionaries; modifying an export does not change the built model.

Projection does not validate or complete a partial runtime model. Supply the required metadata identity and use load_contract() or parse_contract() to validate the exported artifact before using it in a pipeline. For example, include_metadata=False remains useful for runtime validation, but its output lacks the identity required by a complete contract document. Unknown or explicit null document format markers still fail artifact loading.

See Also

• Contract Parser
• Contract Store
• Validator
• Contract-First Guide