Contract Parser

Parse data contract files (YAML/JSON) into ContractMetadata. Promote to the canonical DataContract with metadata.to_data_contract() for use with the Validator, QualityCheck, and ETL.

Overview

from pycharter import parse_contract_file, parse_contract

# From file
metadata = parse_contract_file("user_contract.yaml")

# From dict (canonical key: json_schema)
metadata = parse_contract({"json_schema": {...}, "metadata": {...}})

# Promote to DataContract (for Validator, quality, store)
contract = metadata.to_data_contract()

API Reference

parse_contract

parse_contract(
    contract_data: dict[str, Any], validate: bool = True
) -> ContractMetadata

Parse a contract dictionary and decompose into metadata components.

Expected contract structure: { "json_schema": {...}, # JSON Schema definition (required) "coercion_rules": {...}, # Optional coercion rules (may contain "version") "validation_rules": {...}, # Optional validation rules (may contain "version") "metadata": { # Optional metadata (may contain "version") "ownership": {...}, # Ownership info nested in metadata "governance_rules": {...} # Governance rules nested in metadata }, "field_bindings": {...}, # Optional field bindings artifact "versions": {...} # Optional explicit version tracking }

The contract is validated against Pydantic models to ensure it adheres to the database table design. This ensures data integrity when storing contracts.

Note: ownership and governance_rules should be inside metadata, not at top level.

Parameters:

Name	Type	Description	Default
contract_data	dict[str, Any]	Contract data as dictionary	required
validate	bool	If True (default), validate contract against schema before parsing	True

Returns:

Type	Description
ContractMetadata	ContractMetadata object with decomposed components and version tracking

Raises:

Type	Description
ValueError	If contract does not conform to the required schema (when validate=True)

Example

contract = { ... "json_schema": {"type": "object", "version": "1.0.0", "properties": {"name": {"type": "string"}}}, ... "metadata": { ... "version": "1.0.0", ... "ownership": {"owner": "team-data", "team": "data-engineering"}, ... "governance_rules": {"data_retention": {"days": 365}} ... } ... } metadata = parse_contract(contract) metadata.json_schema {'type': 'object', 'version': '1.0.0', 'properties': {'name': {'type': 'string'}}} metadata.versions {'json_schema': '1.0.0', 'metadata': '1.0.0'} metadata.ownership

Source code in src/pycharter/contract_parser/parser.py

def parse_contract(
    contract_data: dict[str, Any], validate: bool = True
) -> ContractMetadata:
    """
    Parse a contract dictionary and decompose into metadata components.

    Expected contract structure:
    {
        "json_schema": {...},         # JSON Schema definition (required)
        "coercion_rules": {...},      # Optional coercion rules (may contain "version")
        "validation_rules": {...},   # Optional validation rules (may contain "version")
        "metadata": {                 # Optional metadata (may contain "version")
            "ownership": {...},       # Ownership info nested in metadata
            "governance_rules": {...} # Governance rules nested in metadata
        },
        "field_bindings": {...},      # Optional field bindings artifact
        "versions": {...}             # Optional explicit version tracking
    }

    The contract is validated against Pydantic models to ensure it adheres to the
    database table design. This ensures data integrity when storing contracts.

    Note: ownership and governance_rules should be inside metadata, not at top level.

    Args:
        contract_data: Contract data as dictionary
        validate: If True (default), validate contract against schema before parsing

    Returns:
        ContractMetadata object with decomposed components and version tracking

    Raises:
        ValueError: If contract does not conform to the required schema (when validate=True)

    Example:
        >>> contract = {
        ...     "json_schema": {"type": "object", "version": "1.0.0", "properties": {"name": {"type": "string"}}},
        ...     "metadata": {
        ...         "version": "1.0.0",
        ...         "ownership": {"owner": "team-data", "team": "data-engineering"},
        ...         "governance_rules": {"data_retention": {"days": 365}}
        ...     }
        ... }
        >>> metadata = parse_contract(contract)
        >>> metadata.json_schema
        {'type': 'object', 'version': '1.0.0', 'properties': {'name': {'type': 'string'}}}
        >>> metadata.versions
        {'json_schema': '1.0.0', 'metadata': '1.0.0'}
        >>> metadata.ownership
        {'owner': 'team-data', 'team': 'data-engineering'}
    """
    # Native ODCS ingestion (KM-7): an ODCS DataContract document is
    # accepted anywhere a contract is accepted, converted up front.
    # Lazy import — contract_io depends on semantic.field_ref.
    if contract_data.get("kind") == "DataContract" and "apiVersion" in contract_data:
        from pycharter.contract_io.from_odcs import from_odcs

        contract_data = from_odcs(contract_data)
        # ``from_odcs`` retains its public legacy header shape for direct
        # callers. Canonical parsing needs the shared metadata identity.
        contract_data = dict(contract_data)
        metadata = dict(contract_data.get("metadata") or {})
        for key in ("name", "version", "status", "description"):
            value = contract_data.pop(key, None)
            if value is not None:
                metadata.setdefault(key, value)
        contract_data["metadata"] = metadata

    # Upgrade the persisted artifact to the current format on read
    # (DG-2). Stamps an absent ``format_version`` to the current value
    # and raises ``UnsupportedFormatVersion`` for documents written by a
    # newer pycharter. Runs before validation so legacy shapes are
    # upgraded prior to structural checks.
    contract_data = unwrap_manifest(
        "contract", upgrade_artifact_format("contract", contract_data)
    )
    from pycharter.semantic.contracts.rule_artifacts import normalize_rule_artifacts

    contract_data = normalize_rule_artifacts(contract_data)

    for deprecated_key in ("field_mapping", "ontology"):
        if deprecated_key in contract_data:
            raise ValueError(
                f"Retired contract artifact key '{deprecated_key}'. "
                "Use top-level 'field_bindings' instead (breaking release)."
            )

    # Validate contract structure against schema
    if validate:
        _validate_contract_structure(contract_data)

    if "schema" in contract_data:
        raise ValueError(
            "Deprecated contract key 'schema'. Use top-level 'json_schema' instead."
        )
    schema = contract_data.get("json_schema", {})
    coercion_rules = contract_data.get("coercion_rules", {})
    validation_rules = contract_data.get("validation_rules", {})
    metadata = contract_data.get("metadata", {})
    field_bindings = contract_data.get("field_bindings", {})
    data_quality = contract_data.get("data_quality", {})
    sla_default_element = contract_data.get("sla_default_element")
    sla_properties = contract_data.get("sla_properties", [])
    domain_schema = contract_data.get("domain_schema")
    ai_usage_policy = contract_data.get("ai_usage_policy")
    extraction_policy = contract_data.get("extraction_policy")
    kb_capture = contract_data.get("kb_capture")
    # Stamped by the DG-2 upgrade above, which runs before this point, so
    # it is present for every artifact regardless of how the file spelled
    # its envelope.
    format_version = contract_data.get("format_version")
    strict_types = contract_data.get("strict_types")

    # If schema is not at top level, check if entire contract is a schema
    if not schema and ("type" in contract_data or "properties" in contract_data):
        schema = contract_data
        # Extract other components if they exist as separate keys
        coercion_rules = contract_data.get("coercion_rules", {})
        validation_rules = contract_data.get("validation_rules", {})
        metadata = {
            k: v
            for k, v in contract_data.items()
            if k
            not in [
                "json_schema",
                "coercion_rules",
                "validation_rules",
                "versions",
                "field_bindings",
                "data_quality",
                "sla_default_element",
                "sla_properties",
                "domain_schema",
                "ai_usage_policy",
                "extraction_policy",
                "kb_capture",
                "format_version",
            ]
        }

    # Ensure metadata is a dict
    if not isinstance(metadata, dict):
        metadata = {}

    # Move ownership from top level to metadata if present
    if "ownership" in contract_data and "ownership" not in metadata:
        metadata["ownership"] = contract_data.get("ownership")

    # Move governance_rules from top level to metadata if present
    if "governance_rules" in contract_data and "governance_rules" not in metadata:
        metadata["governance_rules"] = contract_data.get("governance_rules")

    # Extract versions from all components
    versions: dict[str, str] = {}

    # Check if explicit versions dict is provided
    if "versions" in contract_data and isinstance(contract_data["versions"], dict):
        explicit_versions = dict(contract_data["versions"])
        if "schema" in explicit_versions:
            raise ValueError(
                "Deprecated versions key 'schema'. Use 'json_schema' instead."
            )
        versions.update(explicit_versions)

    # Extract version from schema
    if isinstance(schema, dict) and "version" in schema:
        versions["json_schema"] = schema["version"]

    # Extract version from metadata
    if isinstance(metadata, dict) and "version" in metadata:
        versions["metadata"] = metadata["version"]

    # Extract version from coercion_rules
    if isinstance(coercion_rules, dict) and "version" in coercion_rules:
        versions["coercion_rules"] = coercion_rules["version"]

    # Extract version from validation_rules
    if isinstance(validation_rules, dict) and "version" in validation_rules:
        versions["validation_rules"] = validation_rules["version"]

    # Extract version from field_bindings
    if isinstance(field_bindings, dict) and "version" in field_bindings:
        versions["field_bindings"] = field_bindings["version"]

    # Extract version from data_quality (DQC)
    if isinstance(data_quality, dict) and "version" in data_quality:
        versions["data_quality"] = data_quality["version"]

    # sla_properties is a list; its version (if present) comes only from
    # the explicit versions footer, which is already merged in above.

    normalized_mapping, mapping_warnings = normalize_field_bindings(field_bindings)
    for message in mapping_warnings:
        warnings.warn(message, UserWarning, stacklevel=2)

    return ContractMetadata(
        json_schema=schema,
        coercion_rules=coercion_rules,
        validation_rules=validation_rules,
        metadata=metadata,
        field_bindings=normalized_mapping,
        versions=versions,
        data_quality=data_quality if isinstance(data_quality, dict) else {},
        sla_default_element=sla_default_element,
        sla_properties=(
            list(sla_properties) if isinstance(sla_properties, list) else []
        ),
        domain_schema=domain_schema,
        ai_usage_policy=(
            ai_usage_policy if isinstance(ai_usage_policy, dict) else None
        ),
        extraction_policy=(
            extraction_policy if isinstance(extraction_policy, dict) else None
        ),
        kb_capture=(kb_capture if isinstance(kb_capture, dict) else None),
        format_version=format_version,
        strict_types=strict_types,
    )

parse_contract_file

parse_contract_file(
    file_path: str, validate: bool = True
) -> ContractMetadata

Load and parse a contract file (YAML or JSON).

The contract is validated against Pydantic models to ensure it adheres to the database table design before parsing.

Parameters:

Name	Type	Description	Default
file_path	str	Path to contract file	required
validate	bool	If True (default), validate contract against schema before parsing	True

Returns:

Type	Description
ContractMetadata	ContractMetadata object with decomposed components

Raises:

Type	Description
FileNotFoundError	If file doesn't exist
ValueError	If file format is not supported, invalid, or doesn't conform to schema

Example

metadata = parse_contract_file("contract.yaml") print(metadata.json_schema)

Source code in src/pycharter/contract_parser/parser.py

def parse_contract_file(file_path: str, validate: bool = True) -> ContractMetadata:
    """
    Load and parse a contract file (YAML or JSON).

    The contract is validated against Pydantic models to ensure it adheres to the
    database table design before parsing.

    Args:
        file_path: Path to contract file
        validate: If True (default), validate contract against schema before parsing

    Returns:
        ContractMetadata object with decomposed components

    Raises:
        FileNotFoundError: If file doesn't exist
        ValueError: If file format is not supported, invalid, or doesn't conform to schema

    Example:
        >>> metadata = parse_contract_file("contract.yaml")
        >>> print(metadata.json_schema)
    """
    path = Path(file_path)

    if not path.exists():
        raise FileNotFoundError(f"Contract file not found: {file_path}")

    # Determine file format
    suffix = path.suffix.lower()

    if suffix in [".yaml", ".yml"]:
        with open(path, encoding="utf-8") as f:
            contract_data = yaml.safe_load(f)
    elif suffix == ".json":
        with open(path, encoding="utf-8") as f:
            contract_data = json.load(f)
    else:
        raise ValueError(
            f"Unsupported file format: {suffix}. Supported formats: .json, .yaml, .yml"
        )

    if not isinstance(contract_data, dict):
        raise ValueError(
            f"Contract file must contain a dictionary/object, got {type(contract_data)}"
        )

    # Resolve variable substitutions in contract data
    from pycharter.utils.value_injector import resolve_values

    contract_data = resolve_values(contract_data, source_file=file_path)

    return parse_contract(contract_data, validate=validate)

ContractMetadata

The result of parsing a contract. Use .to_data_contract() to get a DataContract.

Attribute	Type	Description
json_schema	Dict	JSON Schema definition
coercion_rules	Dict	Coercion rules (defaults to {} if absent)
validation_rules	Dict	Validation rules (defaults to {} if absent)
metadata	Dict	Metadata (title, description, version)
ownership	Dict	Ownership information
governance_rules	Dict	Governance policies
field_bindings	Dict	Field bindings linking schema fields to vocabulary concepts
versions	Dict	Component versions

Examples

from pycharter import parse_contract_file

# Parse from file
metadata = parse_contract_file("contracts/user.yaml")

# Access components
print(metadata.json_schema)
print(metadata.metadata)
print(metadata.ownership)

# Promote to DataContract for validation or storage
contract = metadata.to_data_contract()
validator = Validator.from_dict(contract)

See Also

• Contract Builder — build DataContract from artifacts or store
• Contract-First Guide
• Contracts Tutorial