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(
    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
    "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) -> 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_mapping: 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( ... 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', 'metadata', 'schema', '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_mapping: 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(
        ...     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', 'metadata', 'schema', 'validation_rules']
        >>> "coercion" not in (contract.json_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] = {
        "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.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 field_mapping
    if artifacts.field_mapping:
        mapping_version = _extract_version(artifacts.field_mapping)
        if mapping_version:
            versions["field_mapping"] = 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_mapping if present
    if artifacts.field_mapping:
        contract["field_mapping"] = copy.deepcopy(artifacts.field_mapping)

    # 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) -> 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_mapping).

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`

Returns:

Type	Description
`DataContract`	DataContract ready for runtime validation

Raises:

Type	Description
`ValueError`	If the contract (or its schema) is not found

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,
) -> 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_mapping).

    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

    Returns:
        DataContract ready for runtime validation

    Raises:
        ValueError: If the contract (or its schema) is not found
    """
    contract = store.get_contract(
        contract_name,
        contract_version,
        include_metadata=include_metadata,
        include_ownership=include_ownership,
        include_governance=include_governance,
    )
    if contract is None:
        raise ValueError(
            f"Contract not found: {contract_name!r} @ {contract_version!r}. "
            "Ensure the schema is stored for this contract."
        )
    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
`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
`ontology`	Dict	Ontology (version, fields with concept/definition/relationships)

Examples¶

from pycharter import build_contract, build_contract_from_store, ContractArtifacts, Validator

# Build from artifacts
artifacts = ContractArtifacts(
    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_dict=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 contract type. All builders and the parser produce it; Validator, QualityCheck, and ETL accept it (or its dict form via .to_dict()).

DataContract ¶

Bases: BaseModel

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

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_mapping `class-attribute` `instance-attribute` ¶

field_mapping: dict[str, Any] | None = Field(default=None, description='Semantic field bindings (ontology)')

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