Incremental Contract validation¶

Contract(list[T]) validates or converts one materialized list as a single container boundary. It is the right contract when list identity, complete container shape, or one aggregate input operation matters. It is not a record processor: the complete list already exists, outputs are retained together, and one invalid member fails that container operation.

For generators, database cursors, event iterables, and large ETL jobs, retain Contract(T) and consume items lazily:

contract = Contract(Trade)

for trade in contract.iter_python(database_cursor()):
    persist(trade)

iter_validate() is the strict Python operation corresponding to validate(). iter_python() is the external structural operation corresponding to from_python(). Both return Iterator[T]; neither creates a result list, error list, batch envelope, stream object, or second Contract.

Laziness and retained compilation¶

Calling either method validates static controls and returns an iterator. It does not call iter(source), pull the first item, validate a value, copy the source, or compile a Contract per item. Source access begins with next().

The retained Contract's strict validator is reused directly. External Python input compiles on the first consumed item and the Contract retains that artifact. Every pulled item enters the selected operation exactly once. There is no pre-validation, retry, or speculative operation outside normal Contract behavior.

Each requested result pulls only enough input to yield the next valid T or terminate according to failure policy. Under continuation, several invalid items may be pulled before the next valid result exists; no rejected item or error is retained by Talea after its callback returns.

Strict and external Python modes¶

Use iter_validate(source) when the iterable should already contain valid internal Python values. Identity-preserving behavior remains the same as validate(): a valid Spec, dataclass, mutable container, TypedDict, or represented internal value is returned under its existing strict semantics. Representation loaders do not run.

Use iter_python(source) for external structural values. Mappings can become Specs or dataclasses, TypedDicts are detached, aliases and tagged unions use their normal input rules, and Representation loaders execute once. Primitive semantics stay strict; this API does not add implicit coercion.

The optional policy=ResourcePolicy(...) on iter_python() applies independently to every item. Depth is per item, nodes are per item, and max_errors limits details collected inside one item. A record containing ten field failures is still one invalid source item for stream accounting. Strict iter_validate() does not acquire an external-input policy merely because values arrive through an iterator.

Indexed errors and fail-fast¶

Fail-fast is the default. The first invalid item raises its ordinary ValidationError facts with one zero-based integer prefix:

(4,)
(4, "amount")
(4, "items", 2)

Existing detail ordering, codes, expected/received facts, union branches, related locations, truncation, Sensitive redaction, and permitted causes are preserved by the canonical error-prefix owner. The rejected item is not added merely to locate the error or passed as a separate callback argument. The source is not read past the failing item.

Explicit continuation¶

Continuation requires on_error(index, error). Returning normally says that the application handled that invalid record operationally and permits the iterator to seek the next valid result:

def report_invalid(index: int, error: ValidationError) -> None:
    metrics.increment("invalid_trade")
    logger.warning("trade %d: %s", index, error.errors())


for trade in contract.iter_python(cursor, on_error=report_invalid):
    persist(trade)

The explicit index avoids parsing locations for record identity. The callback also receives the located canonical error, but no separate rejected-value argument. Ordinary non-sensitive ValidationError facts remain available; Sensitive input remains redacted. Talea calls the callback once per invalid source item, not once per nested detail. There is no silent skip option, implicit collection, Result value, retry, predicate, global handler, or callback registry.

Callback exceptions propagate unchanged and stop iteration. Callbacks are trusted synchronous application code and may perform I/O, retain the supplied error, or reenter Talea; Talea holds no lock while invoking them. An event consumer may report and continue, while a payment service may choose fail-fast. That recovery decision belongs to the application.

Stream limits¶

ItemPolicy is available from talea.contract and is immutable:

from talea.contract import ItemPolicy

policy = ItemPolicy(max_items=1_000_000, max_invalid_items=100)

The defaults are finite and deliberately generous. max_items counts every source item pulled, valid or invalid. The count is checked before validation work starts for that item; the first item over the limit raises ResourceLimitError(code="items", limit=N, observed=N + 1). Talea must pull that item to know the source continued, but does not validate it.

max_invalid_items counts invalid source items admitted to continuation. At the first item over the limit, Talea raises ResourceLimitError(code="invalid_items", ...) before calling on_error. Resource failures are control failures, not ordinary item validation failures, so continuation cannot suppress them.

Setting either dimension to None explicitly opts out of that bound. For an infinite iterable, keep a finite max_items or guarantee consumer termination. An invalid infinite iterable with continuation also needs a finite max_invalid_items. Stream policy does not bound source I/O time, callback work, or arbitrary application code.

Source ownership and early termination¶

The caller owns iterable, generator, cursor, file, transaction, retry, and lifetime policy. Talea owns item validation/conversion, index location, failure policy, and its finite limits. Source RuntimeError, OSError, and other application exceptions propagate unchanged because no value was validated at that point.

Talea performs no lookahead, pre-count with len(), drain, retry, or automatic close of an arbitrary source. Closing the returned generator stops Talea consumption and does not swallow GeneratorExit; it does not proactively close the caller's underlying generator. Use the cursor or transaction's context manager when cleanup must be deterministic.

A re-iterable collection creates a fresh iterator on each invocation. Passing the same one-shot iterator to two operations naturally shares caller-owned state. Multiple operations from one Contract have independent indexes and budgets. One iterator is not made thread-safe; separate iterators may be consumed concurrently under normal Python rules.

Memory, mutation, and performance¶

The iterator keeps its active source and current yielded value as ordinary generator state. It does not retain prior successful outputs after advancement, prior continued errors after callback return, or an accumulating history. The source itself may retain values according to its implementation. If the application stores outputs or errors, that retention is application-owned.

Strict mode validates mutable values in their current state. External mode uses normal conversion and yields mutable outputs normally; neither snapshots arbitrary items merely because consumption is incremental.

The permanent benchmark_incremental task measures 1/100/10,000-item primitive paths, Specs, dataclasses, TypedDicts, Representations, tagged and recursive values, indexed failures, sparse/dense continuation, limits, infinite-source stopping, allocation, retention, concurrency, and ordinary Contract/Spec canaries. Incremental overhead consists of generator iteration, enumerate, counter checks, and the retained operation. External mode also creates normal independent ResourcePolicy state per item. Results depend on Python, hardware, source behavior, and item shape.

Security and failure ownership¶

Sensitive protects Talea-owned failure detail after index prefixing. It cannot prevent the source or callback from logging raw application values. Limits protect Talea-owned counts and validation traversal; they do not provide a wall-clock timeout, source/callback sandbox, database cancellation, or async cancellation boundary.

Framing and current limitations¶

This owner consumes synchronous Python items and does not reinterpret UTF-8, BOM handling, blank lines, malformed syntax, or line numbers. The separate JSON Lines input owner performs that framing and strict decode, then composes decoded values and both failure domains with the same internal item/invalid-item accounting truth.

There is no AsyncIterable or async callback support, streaming serialization, JSONL output, automatic retry, silent ignore mode, Result API, wall-clock timeout, or source sandbox. Generic and decoded JSONL validation item indexes are always zero-based.

Complete executable trade workflow¶

This program covers strict and external modes, a generator cursor, fail-fast, continuation, indexed and Sensitive errors, both stream limits, early stop, source failure, and a memory-friendly persistence loop:

"""Executable incremental trade-processing examples."""

from __future__ import annotations

from collections.abc import Generator, Iterator
from typing import Annotated

from talea import Contract, ResourceLimitError, Sensitive, Spec, ValidationError
from talea.contract import ItemPolicy

type AccountToken = Annotated[int, Sensitive()]


class Trade(Spec):
    """One validated trade accepted by the persistence boundary."""

    trade_id: int
    account_token: AccountToken
    quantity: int


trade_contract = Contract(Trade)


def database_cursor() -> Iterator[dict[str, object]]:
    """Stand in for an application-owned database cursor."""

    yield {"trade_id": 1, "account_token": 801, "quantity": 5}
    yield {"trade_id": 2, "account_token": 802, "quantity": 8}


# Iterator creation consumes no source item. Each mapping is pulled, converted,
# persisted, and released before the next one is requested.
converted = trade_contract.iter_python(database_cursor())
assert iter(converted) is converted
assert next(converted).trade_id == 1
assert next(converted).trade_id == 2

# Strict item validation is a separate boundary and preserves object identity.
strict_trade = Trade(trade_id=3, account_token=803, quantity=13)
assert next(trade_contract.iter_validate((strict_trade,))) is strict_trade

# Fail-fast is the default and locations start with the zero-based source index.
try:
    list(
        trade_contract.iter_python(
            (
                {"trade_id": 4, "account_token": 804, "quantity": 3},
                {"trade_id": 5, "account_token": 805, "quantity": "invalid"},
            )
        )
    )
except ValidationError as error:
    assert error.errors()[0]["location"] == [1, "quantity"]
else:
    raise AssertionError("invalid trade should fail fast")

# Continuation is an explicit application recovery decision. The callback gets
# the index and located canonical error, but no separate raw-mapping argument.
reported: list[tuple[int, tuple[object, ...]]] = []


def report_invalid(index: int, error: ValidationError) -> None:
    reported.append((index, tuple(error.errors()[0]["location"])))


accepted = trade_contract.iter_python(
    (
        {"trade_id": 6, "account_token": 806, "quantity": 2},
        {"trade_id": 7, "account_token": 807, "quantity": "invalid"},
        {"trade_id": 8, "account_token": 808, "quantity": 4},
    ),
    on_error=report_invalid,
)
assert [trade.trade_id for trade in accepted] == [6, 8]
assert reported == [(1, (1, "quantity"))]

# Sensitive failure details stay redacted after the index is prefixed.
try:
    next(trade_contract.iter_python(({"trade_id": 9, "account_token": "private-token", "quantity": 1},)))
except ValidationError as error:
    detail = error.errors()[0]
    assert detail["location"] == [0, "account_token"]
    assert detail["input"] == "<redacted>"
    assert "private-token" not in str(error)
else:
    raise AssertionError("sensitive invalid trade should fail")

# Every pulled source item counts, including invalid items. Invalid-item limits
# count source records, not the number of nested validation details.
bounded = trade_contract.iter_python(database_cursor(), item_policy=ItemPolicy(max_items=1))
assert next(bounded).trade_id == 1
try:
    next(bounded)
except ResourceLimitError as error:
    assert (error.code, error.limit, error.observed) == ("items", 1, 2)
else:
    raise AssertionError("record budget should terminate consumption")

invalid_bounded = trade_contract.iter_python(
    (
        {"trade_id": "bad", "account_token": 1, "quantity": 1},
        {"trade_id": "bad", "account_token": 2, "quantity": 2},
    ),
    on_error=lambda _index, _error: None,
    item_policy=ItemPolicy(max_invalid_items=1),
)
try:
    list(invalid_bounded)
except ResourceLimitError as error:
    assert (error.code, error.limit, error.observed) == ("invalid_items", 1, 2)
else:
    raise AssertionError("invalid-record budget should terminate continuation")

# Consumer early termination performs no lookahead or drain.
pulls: list[int] = []


def event_source() -> Iterator[Trade]:
    for index in range(100):
        pulls.append(index)
        yield Trade(trade_id=index, account_token=index, quantity=1)


early = trade_contract.iter_validate(event_source())
assert isinstance(early, Generator)
assert next(early).trade_id == 0
early.close()
assert pulls == [0]

# Source failures remain application exceptions.
source_failure = OSError("cursor disconnected")


def failing_cursor() -> Iterator[dict[str, object]]:
    yield {"trade_id": 10, "account_token": 810, "quantity": 1}
    raise source_failure


failing = trade_contract.iter_python(failing_cursor())
assert next(failing).trade_id == 10
try:
    next(failing)
except OSError as error:
    assert error is source_failure
else:
    raise AssertionError("source exception should propagate unchanged")

# A memory-friendly consumer keeps only application state, not a result batch.
persisted_count = 0
for _trade in trade_contract.iter_python(database_cursor()):
    persisted_count += 1  # persist(trade)
assert persisted_count == 2