Validate

JSON schema validation with streaming support

Classes

Class Description
Validator JSON schema validator with streaming support.

class Validator

Validator(self, schema: type[BaseModel] | dict[str, Any] | str)

JSON schema validator with streaming support.

Validates JSON against a schema, with the unique ability to validate incrementally as data streams in. This enables early abort when schema violations are detected.

Parameters
schema : type[BaseModel] | dict[str, Any] | str
  • The schema to validate against. Can be:
  • A Pydantic model class
  • A JSON schema dictionary
  • A JSON schema string
Examples

Complete validation:

from pydantic import BaseModel
from talu.validate import Validator

class User(BaseModel):
    name: str
    age: int

validator = Validator(User)
validator.validate('{"name":"Alice","age":30}')  # True
validator.validate('{"name":123}')  # False

Streaming validation:

validator = Validator(User)
validator.feed('{"name":')      # True - valid so far
validator.feed('"Alice",')      # True
validator.feed('"age":30}')     # True
validator.is_complete           # True

Early abort on violation:

validator = Validator(User)
validator.feed('{"name":')      # True
validator.feed('123')           # False - abort here!

Raises ------ ValueError If the schema fails to compile.

Quick Reference

Properties

Name Type
error StreamingValidationError | None
is_complete bool
position int

Methods

Method Description
can_continue_with() Check if data could be accepted without advanci...
close() Release native validator resources.
feed() Feed a chunk of data for streaming validation.
reset() Reset validator to initial state.
validate() Validate complete JSON data against the schema.

Properties

error: StreamingValidationError | None

Rich error for the last validation failure.

  • Returns None if validation hasn't failed. After a failed feed(), this property provides detailed diagnostics including:
  • Position where validation failed
  • The invalid byte
  • What bytes would have been valid
  • Context around the failure point
Example
>>> validator.feed('{"name": 123}')  # Returns False
False
>>> if validator.error:
...     print(validator.error.expected)
'"' (quote)

is_complete: bool

True if the data fed so far is valid AND complete.

A complete JSON document has been fully parsed and matches the schema. Use this after feeding all chunks to confirm the document is finished.

position: int

Current byte position in the stream.

Useful for error reporting - indicates where validation stopped or where an error occurred.

Methods

def can_continue_with(self, data: bytes | str) → bool

Check if data could be accepted without advancing state.

Useful for lookahead or testing multiple continuations.

Parameters
data : bytes | str

The data to test.

Returns

bool True if the data would be valid from current state.

def close(self) → None

Release native validator resources.

After calling close(), the validator cannot be used. Safe to call multiple times (idempotent).

def feed(
    self,
    chunk: bytes | str,
    strict: bool = False
) → bool

Feed a chunk of data for streaming validation.

Call repeatedly as data arrives. Returns False immediately if the chunk violates the schema, enabling early abort.

Parameters
chunk : bytes | str

A chunk of JSON data.

strict : bool, optional

If True, raises StreamingValidationError on failure with rich diagnostics (position, invalid byte, expected bytes, context). Default is False for backward compatibility.

Returns

bool True if the chunk was valid, False if schema was violated.

Raises
StreamingValidationError

If strict=True and validation fails.

Examples
>>> validator = Validator(User)
>>> validator.feed('{"name":')       # True
>>> validator.feed('"Alice",')       # True
>>> validator.feed('"age":30}')      # True
>>> validator.is_complete            # True

>>> validator.reset()
>>> validator.feed('{"name":')       # True
>>> validator.feed('123')            # False - number not allowed

With strict mode for immediate error diagnostics:

>>> validator.reset()
>>> validator.feed('{"name":', strict=True)
>>> validator.feed('123', strict=True)
StreamingValidationError: Invalid byte '1' at position 8. Expected: '"' (quote)

def reset(self) → None

Reset validator to initial state.

Call this to reuse the validator for a new validation.

Raises
TaluError

If the validator state cannot be reset.

def validate(self, data: bytes | str) → bool

Validate complete JSON data against the schema.

This is the simple, one-shot validation API. For streaming validation, use feed() instead.

Parameters
data : bytes | str

The complete JSON data to validate.

Returns

bool True if the data is valid according to the schema.

Examples
>>> validator = Validator({"type": "object", "properties": {"x": {"type": "integer"}}})
>>> validator.validate('{"x": 42}')
True
>>> validator.validate('{"x": "not an int"}')
False