Contract Builder¶

Build consolidated contracts from separate artifacts.

Overview¶

from pycharter import build_contract, ContractArtifacts

artifacts = ContractArtifacts(
    schema=schema,
    coercion_rules=coercion,
    validation_rules=validation
)
contract = build_contract(artifacts)

Contract Structure¶

Schema is RAW

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

The returned contract always has this structure:

{
    "schema": {...},           # RAW 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
    "ontology": {...},         # Optional ontology (semantic field annotations)
    "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,
) -> dict[str, Any]

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
	`dict[str, Any]`	Consolidated contract dictionary with:
	`dict[str, Any]`	schema: RAW schema (no coercion/validation merged; must have version)
	`dict[str, Any]`	coercion_rules: Coercion rules dictionary (always present, may be empty)
	`dict[str, Any]`	validation_rules: Validation rules dictionary (always present, may be empty)
	`dict[str, Any]`	metadata: Metadata containing ownership and governance_rules (if include_metadata=True)
	`dict[str, Any]`	ontology: Ontology annotations (if present in artifacts)
	`dict[str, Any]`	versions: Dictionary tracking versions of all components
`Note`	`dict[str, Any]`	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( ... 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) list(contract["versions"].keys()) ['schema', 'coercion_rules', 'validation_rules', 'metadata'] "coercion" not in (contract["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,
) -> dict[str, Any]:
    """
    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:
        Consolidated contract dictionary with:
        - schema: RAW 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)
        - ontology: Ontology annotations (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(
        ...     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)
        >>> list(contract["versions"].keys())
        ['schema', 'coercion_rules', 'validation_rules', 'metadata']
        >>> "coercion" not in (contract["schema"].get("properties") or {}).get("age", {})
        True
    """
    if not artifacts.schema:
        raise ValueError("Schema is required to build a contract")

    # Validate schema has version
    if "version" not in artifacts.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.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] = {
        "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.schema)
    if schema_version:
        versions["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 ontology
    if artifacts.ontology:
        ontology_version = _extract_version(artifacts.ontology)
        if ontology_version:
            versions["ontology"] = ontology_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 ontology if present
    if artifacts.ontology:
        contract["ontology"] = copy.deepcopy(artifacts.ontology)

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

    return contract

build_contract_from_store ¶

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

Build a consolidated data contract from the metadata 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 + ontology).