Skip to content

Validator

The Validator class is the primary interface for data validation.

Overview

from pycharter import Validator

# Create from contract file
validator = Validator.from_file("user_contract.yaml")

# Validate data
result = validator.validate({"name": "Alice", "age": 30})

if result.is_valid:
    print(f"Valid: {result.data}")
else:
    print(f"Errors: {result.errors}")

API Reference

Validator

Validator(
    contract_dir: str | None = None,
    contract_file: str | None = None,
    contract_dict: dict[str, Any] | None = None,
    contract_metadata: ContractMetadata | None = None,
    store: MetadataStoreClient | None = None,
    contract_name: str | None = None,
    contract_version: str | None = None,
)

Generic Validator that performs validation from contract artifacts.

This class can be instantiated with contract files (schema.yaml, coercion_rules.yaml, validation_rules.yaml) or a contract directory, and then used to validate data.

Example

from pycharter.runtime_validator import Validator

Initialize with contract directory

validator = Validator( ... contract_dir="data/examples/fmp_stock_list" ... )

Validate single record

result = validator.validate({"symbol": "AAPL", "company_name": "Apple Inc."}) if result.is_valid: ... print(f"Valid: {result.data}")

Validate batch

results = validator.validate_batch([record1, record2, record3]) valid_count = sum(1 for r in results if r.is_valid)

Using builder pattern

from pycharter.runtime_validator import ValidatorBuilder validator = ( ... ValidatorBuilder() ... .from_directory("data/contracts/user") ... .strict() ... .with_quality_checks() ... .build() ... )

This is the primary and recommended way to perform validation in pycharter. The Validator class handles loading contracts from various sources and generates Pydantic models for validation.

Parameters:

Name Type Description Default
contract_dir str | None

Directory containing contract files (schema.yaml, coercion_rules.yaml, validation_rules.yaml)

None
contract_file str | None

Path to complete contract file (YAML/JSON)

None
contract_dict dict[str, Any] | None

Contract as dictionary with 'schema', 'coercion_rules', 'validation_rules' keys

None
contract_metadata ContractMetadata | None

ContractMetadata object (from parse_contract)

None
store MetadataStoreClient | None

MetadataStoreClient instance (for loading from metadata store)

None
contract_name str | None

Data contract name (required when using store)

None
contract_version str | None

Data contract version (required when using store)

None
Example

From contract directory

validator = Validator(contract_dir="data/contracts/user") result = validator.validate({"name": "Alice", "age": 30})

From metadata store

store = SQLiteMetadataStore("metadata.db") validator = Validator(store=store, contract_name="user_schema", contract_version="1.0") result = validator.validate({"name": "Alice", "age": 30})

Source code in src/pycharter/runtime_validator/validator.py
def __init__(
    self,
    contract_dir: str | None = None,
    contract_file: str | None = None,
    contract_dict: dict[str, Any] | None = None,
    contract_metadata: ContractMetadata | None = None,
    store: MetadataStoreClient | None = None,
    contract_name: str | None = None,
    contract_version: str | None = None,
):
    """
    Initialize the validator with contract artifacts.

    This is the primary and recommended way to perform validation in pycharter.
    The Validator class handles loading contracts from various sources and generates
    Pydantic models for validation.

    Args:
        contract_dir: Directory containing contract files (schema.yaml, coercion_rules.yaml, validation_rules.yaml)
        contract_file: Path to complete contract file (YAML/JSON)
        contract_dict: Contract as dictionary with 'schema', 'coercion_rules', 'validation_rules' keys
        contract_metadata: ContractMetadata object (from parse_contract)
        store: MetadataStoreClient instance (for loading from metadata store)
        contract_name: Data contract name (required when using store)
        contract_version: Data contract version (required when using store)

    Example:
        >>> # From contract directory
        >>> validator = Validator(contract_dir="data/contracts/user")
        >>> result = validator.validate({"name": "Alice", "age": 30})

        >>> # From metadata store
        >>> store = SQLiteMetadataStore("metadata.db")
        >>> validator = Validator(store=store, contract_name="user_schema", contract_version="1.0")
        >>> result = validator.validate({"name": "Alice", "age": 30})
    """
    self.schema = None
    self.coercion_rules = {}
    self.validation_rules = {}
    self.model: Type[BaseModel] | None = None
    self._schema_from_store = (
        False  # Flag to track if schema came from store (already merged)
    )

    # Builder configuration options
    self._strict_mode: bool = False
    self._include_quality: bool = False
    self._quality_thresholds: dict[str, float] | None = None
    self._metrics_collector: "MetricsCollector" | None = None
    self._schema_name: str | None = None
    self._schema_version: str = "1.0.0"

    # Load from various sources (order matters - store takes precedence)
    if store and contract_name and contract_version:
        self._load_from_store(store, contract_name, contract_version)
    elif contract_metadata:
        self._load_from_metadata(contract_metadata)
    elif contract_dict:
        self._load_from_dict(contract_dict)
    elif contract_file:
        self._load_from_file(Path(contract_file))
    elif contract_dir:
        self._load_from_directory(Path(contract_dir))
    else:
        raise ValueError(
            "Must provide one of: contract_dir, contract_file, contract_dict, "
            "contract_metadata, or (store + contract_name + contract_version)"
        )

    # Generate Pydantic model from complete schema
    self._generate_model()

from_file classmethod

from_file(path: str) -> 'Validator'

Create validator from a single contract file.

The file should contain all sections (schema, coercion_rules, validation_rules).

Parameters:

Name Type Description Default
path str

Path to contract file (YAML or JSON)

required

Returns:

Type Description
'Validator'

Configured Validator instance

Example

validator = Validator.from_file("contracts/user_contract.yaml")

Source code in src/pycharter/runtime_validator/validator.py
@classmethod
def from_file(
    cls,
    path: str,
) -> "Validator":
    """
    Create validator from a single contract file.

    The file should contain all sections (schema, coercion_rules, validation_rules).

    Args:
        path: Path to contract file (YAML or JSON)

    Returns:
        Configured Validator instance

    Example:
        validator = Validator.from_file("contracts/user_contract.yaml")
    """
    return cls(contract_file=path)

from_files classmethod

from_files(
    schema: str,
    coercion_rules: str | None = None,
    validation_rules: str | None = None,
) -> "Validator"

Create validator from explicit file paths.

Use any file names you want - no assumptions about naming conventions.

Parameters:

Name Type Description Default
schema str

Path to schema file (YAML or JSON)

required
coercion_rules str | None

Optional path to coercion rules file

None
validation_rules str | None

Optional path to validation rules file

None

Returns:

Type Description
'Validator'

Configured Validator instance

Example

validator = Validator.from_files( schema="my_user_schema.yaml", coercion_rules="type_conversions.yaml", validation_rules="business_rules.yaml" )

Source code in src/pycharter/runtime_validator/validator.py
@classmethod
def from_files(
    cls,
    schema: str,
    coercion_rules: str | None = None,
    validation_rules: str | None = None,
) -> "Validator":
    """
    Create validator from explicit file paths.

    Use any file names you want - no assumptions about naming conventions.

    Args:
        schema: Path to schema file (YAML or JSON)
        coercion_rules: Optional path to coercion rules file
        validation_rules: Optional path to validation rules file

    Returns:
        Configured Validator instance

    Example:
        validator = Validator.from_files(
            schema="my_user_schema.yaml",
            coercion_rules="type_conversions.yaml",
            validation_rules="business_rules.yaml"
        )
    """
    schema_path = Path(schema)
    if not schema_path.exists():
        raise FileNotFoundError(f"Schema file not found: {schema_path}")

    with open(schema_path) as f:
        schema_data = yaml.safe_load(f)

    coercion_data = {}
    if coercion_rules:
        coercion_path = Path(coercion_rules)
        if coercion_path.exists():
            with open(coercion_path) as f:
                coercion_data = yaml.safe_load(f) or {}

    validation_data = {}
    if validation_rules:
        validation_path = Path(validation_rules)
        if validation_path.exists():
            with open(validation_path) as f:
                validation_data = yaml.safe_load(f) or {}

    return cls(
        contract_dict={
            "schema": schema_data,
            "coercion_rules": coercion_data,
            "validation_rules": validation_data,
        }
    )

from_dir classmethod

from_dir(directory: str) -> 'Validator'

Create validator from a directory containing config files.

Expects files with standard names: - schema.yaml (required) - coercion_rules.yaml (optional) - validation_rules.yaml (optional)

Parameters:

Name Type Description Default
directory str

Path to directory containing config files

required

Returns:

Type Description
'Validator'

Configured Validator instance

Example

validator = Validator.from_dir("contracts/users/")

Source code in src/pycharter/runtime_validator/validator.py
@classmethod
def from_dir(
    cls,
    directory: str,
) -> "Validator":
    """
    Create validator from a directory containing config files.

    Expects files with standard names:
    - schema.yaml (required)
    - coercion_rules.yaml (optional)
    - validation_rules.yaml (optional)

    Args:
        directory: Path to directory containing config files

    Returns:
        Configured Validator instance

    Example:
        validator = Validator.from_dir("contracts/users/")
    """
    return cls(contract_dir=directory)

from_dict classmethod

from_dict(
    schema: dict[str, Any],
    coercion_rules: dict[str, Any] | None = None,
    validation_rules: dict[str, Any] | None = None,
) -> "Validator"

Create validator from dictionaries.

Parameters:

Name Type Description Default
schema dict[str, Any]

Schema dictionary (JSON Schema format)

required
coercion_rules dict[str, Any] | None

Optional coercion rules dictionary

None
validation_rules dict[str, Any] | None

Optional validation rules dictionary

None

Returns:

Type Description
'Validator'

Configured Validator instance

Example

validator = Validator.from_dict( schema={ "type": "object", "properties": { "name": {"type": "string"}, "age": {"type": "integer"} } }, coercion_rules={"age": "int"}, validation_rules={"age": {"min": 0, "max": 150}} )

Source code in src/pycharter/runtime_validator/validator.py
@classmethod
def from_dict(
    cls,
    schema: dict[str, Any],
    coercion_rules: dict[str, Any] | None = None,
    validation_rules: dict[str, Any] | None = None,
) -> "Validator":
    """
    Create validator from dictionaries.

    Args:
        schema: Schema dictionary (JSON Schema format)
        coercion_rules: Optional coercion rules dictionary
        validation_rules: Optional validation rules dictionary

    Returns:
        Configured Validator instance

    Example:
        validator = Validator.from_dict(
            schema={
                "type": "object",
                "properties": {
                    "name": {"type": "string"},
                    "age": {"type": "integer"}
                }
            },
            coercion_rules={"age": "int"},
            validation_rules={"age": {"min": 0, "max": 150}}
        )
    """
    return cls(
        contract_dict={
            "schema": schema,
            "coercion_rules": coercion_rules or {},
            "validation_rules": validation_rules or {},
        }
    )

validate

validate(
    data: dict[str, Any],
    strict: bool | None = None,
    include_quality: bool | None = None,
) -> ValidationResult

Validate a single data record against the contract.

Parameters:

Name Type Description Default
data dict[str, Any]

Data dictionary to validate

required
strict bool | None

If True, raise exceptions on validation errors. Defaults to builder setting or False.

None
include_quality bool | None

If True, include quality metrics in result. Defaults to builder setting or False.

None

Returns:

Type Description
ValidationResult

ValidationResult object

Example

validator = Validator(contract_dir="data/examples/fmp_stock_list") result = validator.validate({"symbol": "AAPL", "company_name": "Apple Inc."}) if result.is_valid: ... print(f"Valid: {result.data.symbol}")

Source code in src/pycharter/runtime_validator/validator.py
def validate(
    self,
    data: dict[str, Any],
    strict: bool | None = None,
    include_quality: bool | None = None,
) -> ValidationResult:
    """
    Validate a single data record against the contract.

    Args:
        data: Data dictionary to validate
        strict: If True, raise exceptions on validation errors.
               Defaults to builder setting or False.
        include_quality: If True, include quality metrics in result.
                        Defaults to builder setting or False.

    Returns:
        ValidationResult object

    Example:
        >>> validator = Validator(contract_dir="data/examples/fmp_stock_list")
        >>> result = validator.validate({"symbol": "AAPL", "company_name": "Apple Inc."})
        >>> if result.is_valid:
        ...     print(f"Valid: {result.data.symbol}")
    """
    if not self.model:
        raise ValueError("Model not initialized. Check contract loading.")

    # Use builder settings as defaults
    use_strict = strict if strict is not None else self._strict_mode
    use_quality = (
        include_quality if include_quality is not None else self._include_quality
    )

    # Track timing for metrics
    start_time = time.perf_counter()
    result = validate(self.model, data, strict=use_strict)
    duration_ms = (time.perf_counter() - start_time) * 1000

    # Add quality metrics if requested
    if use_quality:
        result.quality = self._compute_quality_metrics([data], [result])

    # Record metrics if collector is configured
    if self._metrics_collector:
        self._record_metrics(result, duration_ms)

    return result

validate_batch

validate_batch(
    data_list: list[dict[str, Any]],
    strict: bool | None = None,
    include_quality: bool | None = None,
) -> list[ValidationResult]

Validate a batch of data records against the contract.

Parameters:

Name Type Description Default
data_list list[dict[str, Any]]

List of data dictionaries to validate

required
strict bool | None

If True, stop on first validation error. Defaults to builder setting or False.

None
include_quality bool | None

If True, include quality metrics in results. Defaults to builder setting or False.

None

Returns:

Type Description
list[ValidationResult]

List of ValidationResult objects

Example

validator = Validator(contract_dir="data/examples/fmp_stock_list") results = validator.validate_batch([record1, record2, record3]) valid_count = sum(1 for r in results if r.is_valid) invalid_count = sum(1 for r in results if not r.is_valid)

Source code in src/pycharter/runtime_validator/validator.py
def validate_batch(
    self,
    data_list: list[dict[str, Any]],
    strict: bool | None = None,
    include_quality: bool | None = None,
) -> list[ValidationResult]:
    """
    Validate a batch of data records against the contract.

    Args:
        data_list: List of data dictionaries to validate
        strict: If True, stop on first validation error.
               Defaults to builder setting or False.
        include_quality: If True, include quality metrics in results.
                        Defaults to builder setting or False.

    Returns:
        List of ValidationResult objects

    Example:
        >>> validator = Validator(contract_dir="data/examples/fmp_stock_list")
        >>> results = validator.validate_batch([record1, record2, record3])
        >>> valid_count = sum(1 for r in results if r.is_valid)
        >>> invalid_count = sum(1 for r in results if not r.is_valid)
    """
    if not self.model:
        raise ValueError("Model not initialized. Check contract loading.")

    # Use builder settings as defaults
    use_strict = strict if strict is not None else self._strict_mode
    use_quality = (
        include_quality if include_quality is not None else self._include_quality
    )

    # Track timing for metrics
    start_time = time.perf_counter()
    results = validate_batch(self.model, data_list, strict=use_strict)
    duration_ms = (time.perf_counter() - start_time) * 1000

    # Add quality metrics if requested
    if use_quality:
        quality = self._compute_quality_metrics(data_list, results)
        # Add to last result for batch summary
        if results:
            results[-1].quality = quality

    # Record metrics if collector is configured
    if self._metrics_collector:
        self._record_batch_metrics(results, duration_ms)

    return results

get_model

get_model() -> Type[BaseModel]

Get the generated Pydantic model.

Returns:

Type Description
Type[BaseModel]

Pydantic model class

Example

validator = Validator(contract_dir="data/examples/fmp_stock_list") Model = validator.get_model()

Use model directly

instance = Model(symbol="AAPL", company_name="Apple Inc.")

Source code in src/pycharter/runtime_validator/validator.py
def get_model(self) -> Type[BaseModel]:
    """
    Get the generated Pydantic model.

    Returns:
        Pydantic model class

    Example:
        >>> validator = Validator(contract_dir="data/examples/fmp_stock_list")
        >>> Model = validator.get_model()
        >>> # Use model directly
        >>> instance = Model(symbol="AAPL", company_name="Apple Inc.")
    """
    if not self.model:
        raise ValueError("Model not initialized. Check contract loading.")

    return self.model

Factory Methods

from_file

Create validator from a single contract file:

validator = Validator.from_file("contracts/user.yaml")

from_files

Create from separate files:

validator = Validator.from_files(
    schema="schemas/user.yaml",
    coercion_rules="rules/coercion.yaml",
    validation_rules="rules/validation.yaml"
)

from_dir

Create from a directory containing schema.yaml, coercion_rules.yaml, validation_rules.yaml:

validator = Validator.from_dir("contracts/user/")

from_dict

Create from dictionaries:

validator = Validator.from_dict(
    schema={"type": "object", "properties": {...}},
    coercion_rules={"rules": {...}},
    validation_rules={"rules": {...}}
)

From Metadata Store

from pycharter import SQLiteMetadataStore

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

validator = Validator(store=store, schema_id="user_schema_v1")

ValidationResult

ValidationResult

ValidationResult(
    is_valid: bool,
    data: BaseModel | None = None,
    errors: list[dict[str, Any]] | None = None,
    quality: QualityMetrics | None = None,
    state: str | None = None,
    trigger: str | None = None,
    entity_id: str | None = None,
    contract_name: str | None = None,
)

Result of a validation operation.

Attributes:

Name Type Description
is_valid

Whether validation passed

data

Validated data (Pydantic model instance) if valid

errors

List of validation errors if invalid

quality

Optional quality metrics

Source code in src/pycharter/runtime_validator/validator_core.py
def __init__(
    self,
    is_valid: bool,
    data: BaseModel | None = None,
    errors: list[dict[str, Any]] | None = None,
    quality: QualityMetrics | None = None,
    # Lifecycle context — set by caller (e.g. bridge code), not by validator
    state: str | None = None,
    trigger: str | None = None,
    entity_id: str | None = None,
    contract_name: str | None = None,
):
    self.is_valid = is_valid
    self.data = data
    self.errors = errors or []
    self.quality = quality
    self.state = state
    self.trigger = trigger
    self.entity_id = entity_id
    self.contract_name = contract_name

is_valid instance-attribute

is_valid = is_valid

data instance-attribute

data = data

errors instance-attribute

errors = errors or []

Examples

Single Validation

result = validator.validate({
    "name": "Alice",
    "email": "alice@example.com",
    "age": 30
})

if result.is_valid:
    user = result.data  # Pydantic model instance
    print(user.name)
else:
    for error in result.errors:
        print(f"{error['loc']}: {error['msg']}")

Batch Validation

records = [
    {"name": "Alice", "email": "alice@example.com"},
    {"name": "", "email": "invalid"},  # Invalid
    {"name": "Charlie", "email": "charlie@example.com"},
]

results = validator.validate_batch(records)

valid_records = [r.data for r in results if r.is_valid]
print(f"Valid: {len(valid_records)}/{len(records)}")

Strict Mode

from pydantic import ValidationError

try:
    result = validator.validate(data, strict=True)
    # If we get here, data is valid
except ValidationError as e:
    print(f"Validation failed: {e}")

Getting the Model

# Access the generated Pydantic model
UserModel = validator.get_model()

# Use it directly
user = UserModel(name="Alice", email="alice@example.com")

# Export schema
print(UserModel.model_json_schema())

Convenience Functions

validate_with_contract

Quick validation without creating a Validator instance:

from pycharter import validate_with_contract

result = validate_with_contract("contract.yaml", data)

validate_with_store

Validate using a schema from the metadata store:

from pycharter import validate_with_store

result = validate_with_store(store, "user_schema_v1", data)

See Also