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: ContractMetadata
    | dict[str, Any]
    | DataContract,
    *,
    strict: bool = False,
    coerce: bool = True,
    strict_types: bool | None = None,
    include_quality: bool = False,
    metrics: MetricsCollector | None = None,
    state_rules: dict[str, Any] | None = None,
)

Generic Validator that performs validation from contract artifacts.

Construct directly from an in-memory contract (a ContractMetadata, a contract dict, or a :class:DataContract), or use one of the from_* classmethods (from_file, from_dir, from_dict, from_store). Then call validate() or validate_batch() on data dicts.

Thread safety

A Validator is immutable after construction — the Pydantic model is compiled once in __init__ and never mutated by validation calls. validate(), validate_batch(), and validate_for_state() are safe to call concurrently from multiple threads on one shared instance. One caveat: an attached metrics collector receives results from every thread, so its store must itself be thread-safe (the bundled InMemoryMetricsStore is).

Parameters:

Name	Type	Description	Default
contract	ContractMetadata | dict[str, Any] | DataContract	The contract source — a ContractMetadata, a contract dict (full artifact dict or a bare JSON Schema document), or a :class:DataContract.	required
strict	bool	If True, validation errors raise instead of being captured in the result. This governs error handling, not typing — see strict_types for the latter. (The name predates Pydantic's own strict; they are unrelated.)	False
coerce	bool	If True (default), the contract's coercion_rules are compiled once into a :class:~pycharter.semantic.contracts.coercion.CoercionPlan and applied to every record before the model runs — the single coercion execution path (models never bake coercion). False skips those declared rules; it does not stop Pydantic from converting types, which is what strict_types controls.	True
strict_types	bool | None	If True, the generated model runs in Pydantic's strict mode, so a value of the wrong JSON type is rejected instead of converted — "10.5" fails a number field rather than becoming 10.5. None (default) takes the contract's own strict_types flag, falling back to False. Leave it off for ingest, where a CSV's "10.5" really is that float and rejecting it would be wrong. Turn it on for verification, where the producer emitted the wrong type and the point of validating is to find out.	None
include_quality	bool	If True, validation results include quality metrics by default.	False
metrics	MetricsCollector | None	Optional MetricsCollector for recording metrics of every validation run.	None
state_rules	dict[str, Any] | None	Optional mapping from state names to override dicts, consumed by validate_for_state().	None

Raises:

Type	Description
CoercionOperationNotFoundError	If coerce=True and the contract names an unknown coercion operation (fail-fast for store-loaded contracts that bypass the load check).

Source code in src/pycharter/runtime_validator/_engine.py

def __init__(
    self,
    contract: ContractMetadata | dict[str, Any] | DataContract,
    *,
    strict: bool = False,
    coerce: bool = True,
    strict_types: bool | None = None,
    include_quality: bool = False,
    metrics: MetricsCollector | None = None,
    state_rules: dict[str, Any] | None = None,
) -> None:
    """Initialize the validator from an in-memory contract.

    Args:
        contract: The contract source — a ``ContractMetadata``, a
            contract ``dict`` (full artifact dict or a bare JSON
            Schema document), or a :class:`DataContract`.
        strict: If True, validation errors raise instead of being
            captured in the result. This governs *error handling*, not
            typing — see ``strict_types`` for the latter. (The name
            predates Pydantic's own ``strict``; they are unrelated.)
        coerce: If True (default), the contract's ``coercion_rules``
            are compiled once into a
            :class:`~pycharter.semantic.contracts.coercion.CoercionPlan`
            and applied to every record *before* the model runs —
            the single coercion execution path (models never bake
            coercion). ``False`` skips those declared rules; it does
            **not** stop Pydantic from converting types, which is what
            ``strict_types`` controls.
        strict_types: If True, the generated model runs in Pydantic's
            strict mode, so a value of the wrong JSON type is rejected
            instead of converted — ``"10.5"`` fails a ``number`` field
            rather than becoming ``10.5``. ``None`` (default) takes the
            contract's own ``strict_types`` flag, falling back to False.

            Leave it off for **ingest**, where a CSV's ``"10.5"`` really
            is that float and rejecting it would be wrong. Turn it on for
            **verification**, where the producer emitted the wrong type
            and the point of validating is to find out.
        include_quality: If True, validation results include quality
            metrics by default.
        metrics: Optional ``MetricsCollector`` for recording metrics
            of every validation run.
        state_rules: Optional mapping from state names to override
            dicts, consumed by ``validate_for_state()``.

    Raises:
        CoercionOperationNotFoundError: If ``coerce=True`` and the
            contract names an unknown coercion operation (fail-fast
            for store-loaded contracts that bypass the load check).
    """
    self.schema: dict[str, Any] | None = None
    self.coercion_rules: dict[str, Any] = {}
    self.validation_rules: dict[str, Any] = {}
    # Batch-level rules (row_count, null rates, …); evaluated over a
    # whole batch by ``validate_records``, never per record.
    self.data_quality: dict[str, Any] = {}
    self._model: type[BaseModel] | None = None
    # JSON Schema assertions the model does not enforce, evaluated by a
    # standards engine after the model; None when the model covers all.
    self._residual_checker: ResidualChecker | None = None
    # ``{field: [(assertion, callable), ...]}`` compiled from
    # warning-severity validation rules; empty when none declared.
    self._warning_checkers: dict[str, Any] = {}
    # Contract-flagged ``pii`` / ``restricted`` fields whose values
    # must never appear in a diagnostic; empty for every contract
    # without ``field_bindings`` flags, which keeps messages byte-
    # identical to previous releases.
    self._withhold_fields: frozenset[str] = frozenset()

    # Configuration options
    self._strict_mode: bool = strict
    # None until the contract is loaded: an explicit argument wins, and
    # the contract's own flag is the fallback.
    self._strict_types_override: bool | None = strict_types
    self._contract_strict_types: bool = False
    self._include_quality: bool = include_quality
    self._quality_thresholds: dict[str, float] | None = None
    self._metrics_collector: MetricsCollector | None = metrics
    self._schema_version: str = "1.0.0"
    if state_rules is not None:
        self._state_rules: dict[str, Any] = state_rules

    self._load_contract(contract)
    self._coercion_plan: CoercionPlan = (
        compile_coercion(self.coercion_rules) if coerce else CoercionPlan(())
    )
    self._generate_model()

model property

model: type[BaseModel]

The generated Pydantic model.

Raises:

Type	Description
ValueError	If model not initialized.

from_file classmethod

from_file(path: str | Path, **kwargs: Any) -> Validator

Create a validator from a single contract file.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to a contract file (YAML/JSON).	required
**kwargs	Any	Forwarded to __init__.	{}

Returns:

Type	Description
Validator	A configured :class:Validator.

Source code in src/pycharter/runtime_validator/_engine.py

@classmethod
def from_file(cls, path: str | Path, **kwargs: Any) -> Validator:
    """Create a validator from a single contract file.

    Args:
        path: Path to a contract file (YAML/JSON).
        **kwargs: Forwarded to ``__init__``.

    Returns:
        A configured :class:`Validator`.
    """
    from pycharter.contract_parser import parse_contract_file

    contract = parse_contract_file(str(path))
    return cls(contract, **kwargs)

from_dir classmethod

from_dir(path: str | Path, **kwargs: Any) -> Validator

Create a validator from a directory of contract files.

Parameters:

Name	Type	Description	Default
path	str | Path	Directory containing schema.yaml (plus optional coercion_rules.yaml / validation_rules.yaml).	required
**kwargs	Any	Forwarded to __init__ (strict, include_quality, metrics, state_rules).	{}

Returns:

Type	Description
Validator	A configured :class:Validator.

Source code in src/pycharter/runtime_validator/_engine.py

@classmethod
def from_dir(cls, path: str | Path, **kwargs: Any) -> Validator:
    """Create a validator from a directory of contract files.

    Args:
        path: Directory containing ``schema.yaml`` (plus optional
            ``coercion_rules.yaml`` / ``validation_rules.yaml``).
        **kwargs: Forwarded to ``__init__`` (``strict``,
            ``include_quality``, ``metrics``, ``state_rules``).

    Returns:
        A configured :class:`Validator`.
    """
    schema, coercion, validation = load_from_directory(Path(path))
    contract = {
        "json_schema": schema,
        "coercion_rules": coercion,
        "validation_rules": validation,
    }
    return cls(contract, **kwargs)

from_dict classmethod

from_dict(
    config: dict[str, Any], **kwargs: Any
) -> Validator

Create a validator from a contract dictionary.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Full artifact dict (json_schema, coercion_rules, validation_rules) or a bare JSON Schema document.	required
**kwargs	Any	Forwarded to __init__.	{}

Returns:

Type	Description
Validator	A configured :class:Validator.

Source code in src/pycharter/runtime_validator/_engine.py

@classmethod
def from_dict(cls, config: dict[str, Any], **kwargs: Any) -> Validator:
    """Create a validator from a contract dictionary.

    Args:
        config: Full artifact dict (``json_schema``, ``coercion_rules``,
            ``validation_rules``) or a bare JSON Schema document.
        **kwargs: Forwarded to ``__init__``.

    Returns:
        A configured :class:`Validator`.
    """
    return cls(config, **kwargs)

from_store classmethod

from_store(
    ref: StoreRef | str,
    *,
    store: ContractStoreClient,
    namespace: str | None = None,
    **kwargs: Any,
) -> Validator

Create a validator from a contract in a contract store.

Parameters:

Name	Type	Description	Default
ref	StoreRef | str	A :class:StoreRef or "name@version" / "name" string addressing the contract in the store.	required
store	ContractStoreClient	The :class:ContractStoreClient to load from.	required
namespace	str | None	Exact contract namespace. None keeps legacy name-only resolution, including ambiguity errors.	None
**kwargs	Any	Forwarded to __init__.	{}

Returns:

Type	Description
Validator	A configured :class:Validator.

Raises:

Type	Description
ValueError	If the contract is not found in the store.

Source code in src/pycharter/runtime_validator/_engine.py

@classmethod
def from_store(
    cls,
    ref: StoreRef | str,
    *,
    store: ContractStoreClient,
    namespace: str | None = None,
    **kwargs: Any,
) -> Validator:
    """Create a validator from a contract in a contract store.

    Args:
        ref: A :class:`StoreRef` or ``"name@version"`` / ``"name"``
            string addressing the contract in the store.
        store: The :class:`ContractStoreClient` to load from.
        namespace: Exact contract namespace. ``None`` keeps legacy
            name-only resolution, including ambiguity errors.
        **kwargs: Forwarded to ``__init__``.

    Returns:
        A configured :class:`Validator`.

    Raises:
        ValueError: If the contract is not found in the store.
    """
    from pycharter.shared.refs import StoreRef

    sr = StoreRef.parse(ref)
    # The envelope document is the full contract (strict_types, privacy
    # flags, data_quality, …), not just the schema and rules.
    envelope = (
        store.get(sr.name, sr.version, namespace=namespace)
        if namespace is not None
        else store.get(sr.name, sr.version)
    )
    document = envelope.get("document") if envelope else None
    if not document or not document.get("json_schema"):
        raise ValueError(
            f"Contract {sr.name!r} @ {sr.version or 'latest'!r} not found in store"
        )
    return cls(document, **kwargs)

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 the configured setting or False.	None
include_quality	bool | None	If True, include quality metrics in result. Defaults to the configured setting or False.	None

Returns:

Type	Description
ValidationResult	ValidationResult object.

Source code in src/pycharter/runtime_validator/_engine.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 the configured setting or False.
        include_quality: If True, include quality metrics in result.
            Defaults to the configured setting or False.

    Returns:
        ValidationResult object.
    """
    if not self._model:
        raise ValueError("Model not initialized. Check contract loading.")

    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
    )

    start_time = time.perf_counter()
    data, coercion_failure = self._run_coercion(data, use_strict)
    result = (
        coercion_failure
        if coercion_failure is not None
        else self._validate_model(self._model, data, use_strict)
    )
    duration_ms = (time.perf_counter() - start_time) * 1000

    if self._warning_checkers:
        result.diagnostics.extend(
            withhold_diagnostic_values(
                check_warning_rules(self._warning_checkers, data),
                self._withhold_fields,
            )
        )

    if use_quality:
        result.quality = compute_quality_metrics([data], [result])

    if self._metrics_collector:
        record_metrics(
            self._metrics_collector,
            result,
            duration_ms,
            self._db_schema or "unknown",
            self._schema_version,
        )

    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 the configured setting or False.	None
include_quality	bool | None	If True, include quality metrics in results. Defaults to the configured setting or False.	None

Returns:

Type	Description
list[ValidationResult]	List of ValidationResult objects.

Source code in src/pycharter/runtime_validator/_engine.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 the configured setting or False.
        include_quality: If True, include quality metrics in results.
            Defaults to the configured setting or False.

    Returns:
        List of ValidationResult objects.
    """
    if not self._model:
        raise ValueError("Model not initialized. Check contract loading.")

    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
    )

    start_time = time.perf_counter()
    processed: list[dict[str, Any]] = []
    results: list[ValidationResult] = []
    for index, record in enumerate(data_list):
        coerced, failure = self._run_coercion(record, use_strict, row=index)
        processed.append(coerced)
        # A coercion failure is the record's primary error; the model
        # would only grade the raw value it could not convert.
        results.append(
            failure
            if failure is not None
            else self._validate_model(self._model, coerced, use_strict, row=index)
        )
    duration_ms = (time.perf_counter() - start_time) * 1000

    if self._warning_checkers:
        for index, (record, result) in enumerate(
            zip(processed, results, strict=False)
        ):
            result.diagnostics.extend(
                withhold_diagnostic_values(
                    check_warning_rules(self._warning_checkers, record, row=index),
                    self._withhold_fields,
                )
            )

    if use_quality:
        # Per-record quality, matching validate(): each result carries its
        # own metrics. (Previously the batch aggregate was attached to the
        # last result only, so every other result silently read None.)
        for record, result in zip(processed, results, strict=False):
            result.quality = compute_quality_metrics([record], [result])

    if self._metrics_collector:
        record_batch_metrics(
            self._metrics_collector,
            results,
            duration_ms,
            self._db_schema or "unknown",
            self._schema_version,
        )

    return results

json_schema

json_schema() -> dict[str, Any]

Return the complete schema (with rules merged).

Returns:

Type	Description
dict[str, Any]	Complete schema dictionary.

Source code in src/pycharter/runtime_validator/_engine.py

def json_schema(self) -> dict[str, Any]:
    """Return the complete schema (with rules merged).

    Returns:
        Complete schema dictionary.
    """
    return self._merge_rules_into_schema()

Constructor and Factory Methods

Direct construction

Construct directly from an in-memory contract — a DataContract, a contract dict, or a ContractMetadata:

from pycharter import Validator, build_contract_from_store

contract = build_contract_from_store(store, "user", "1.0.0")
validator = Validator(contract)  # DataContract, dict, or ContractMetadata
result = validator.validate(data)

Configuration options are keyword-only:

validator = Validator(
    contract,
    strict=True,
    strict_types=True,
    include_quality=True,
)

strict and strict_types are different things

strict controls error handling — whether a validation failure raises instead of being returned in the result. It predates Pydantic's own strict and is unrelated to it.

strict_types controls typing — whether a value of the wrong JSON type is rejected instead of converted.

coerce is a third, separate thing: it toggles the contract's declared coercion_rules. Setting coerce=False does not stop Pydantic from converting types.

from_file

Create validator from a single contract file:

validator = Validator.from_file("contracts/user.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 a contract dictionary (full artifact dict or a bare JSON Schema document):

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

from_store

Create a validator from a contract stored by name and version, addressed with a StoreRef:

from pycharter import Validator, PostgresContractStore
from pycharter.shared.refs import StoreRef

store = PostgresContractStore("postgresql://...")
store.connect()

validator = Validator.from_store(StoreRef("user", "1.0.0"), store=store)
result = validator.validate(data)
store.disconnect()

ValidationResult

ValidationResult

ValidationResult(
    is_valid: bool,
    data: BaseModel | None = None,
    errors: list[str] | None = None,
    quality: QualityMetrics | None = None,
    diagnostics: list[FieldDiagnostic] | 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 human-readable validation error messages if invalid
diagnostics		Structured per-field diagnostics 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[str] | None = None,
    quality: QualityMetrics | None = None,
    diagnostics: list[FieldDiagnostic] | 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.diagnostics = diagnostics 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 (raise instead of return)

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}")

Strict Types (reject wrong types instead of converting)

By default a contract validates leniently: "10.5" satisfies a number field, because Pydantic converts it. That is the right behaviour for ingest — a CSV's "10.5" really is that float, and rejecting it would be wrong.

It is the wrong behaviour for verification, where the producer emitting a string instead of a number is exactly what you are validating to catch. Turn on strict_types for that case:

lenient = Validator(contract)
lenient.validate({"total": "10.5"}).is_valid       # True — converted

strict = Validator(contract, strict_types=True)
strict.validate({"total": "10.5"}).is_valid        # False — float_type
strict.validate({"total": 10.5}).is_valid          # True

It applies to nested objects and array items too, so a nested field cannot slip through while the record it belongs to is being graded strictly.

Declare it on the contract when strictness is a property of the data rather than of one call site — whether a field tolerates "10.5" usually belongs to the contract, not to whoever happens to be validating against it:

format_version: "1"
strict_types: true
json_schema:
  version: "1.0.0"
  type: object
  properties:
    total: { type: number }
  required: [total]

A strict_types argument overrides the contract's flag; absent from both, the lenient default applies, so existing contracts are unaffected.

Getting the Model

# Access the generated Pydantic model
UserModel = validator.model

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

# Export schema
print(UserModel.model_json_schema())

Convenience Functions

get_model_from_contract

Generate a Pydantic model from a contract (file path, DataContract, dict, or ContractMetadata) without keeping a Validator around:

from pycharter import get_model_from_contract

UserModel = get_model_from_contract("contract.yaml")
UserModel = get_model_from_contract(contract_dict)  # DataContract or dict

get_model_from_store

Generate a Pydantic model from a contract in the contract store (by name and version):

from pycharter import get_model_from_store

UserModel = get_model_from_store(store, "user", "1.0.0")

See Also

• Validate a pandas DataFrame — validate_dataframe(df, contract, on_error=...), the one-call façade over Validator.validate_batch with quarantine/fail/warn behaviour.
• Validate a file from the shell — the pycharter validate CLI verb (exit-code CI gate).
• Contracts Tutorial
• Custom Validators Guide
• Pydantic Generator
• Glossary