Positional NamedTuple contracts¶

typing.NamedTuple is useful when a Python application and its wire protocol agree that field order is part of the contract. Talea preserves that choice: the Python value keeps its nominal NamedTuple type, while every external and serialized representation is positional.

from typing import NamedTuple

from talea import Contract


class Coordinate(NamedTuple):
    latitude: float
    longitude: float


coordinates = Contract(Coordinate)
value = coordinates.from_json("[47.37,8.54]")

assert type(value) is Coordinate
assert coordinates.to_python(value) == (47.37, 8.54)

{"latitude": 47.37, "longitude": 8.54} is not an alternative representation. Choose a Spec, stdlib dataclass, or TypedDict when names belong on the external boundary.

Boundary semantics¶

Operation	NamedTuple behavior
`validate()`	requires the exact declared type and complete declared arity, validates each stored slot, preserves identity
`from_python()`	accepts exact `list` or exact `tuple`, converts slots, invokes the declared constructor once
`from_json()`	accepts a JSON array after strict JSON decoding
`to_python()`	returns an ordinary tuple containing recursively projected values
`to_json()`	returns a JSON array
`json_schema()`	projects a named Draft 2020-12 array definition
`openapi_schema()`	reuses the same array Schema Object under OpenAPI components

A plain tuple or list fails strict validation even when its contents match. External conversion accepts only the two concrete built-in positional families; strings, bytes, mappings, custom sequences, and tuple/list subclasses are not silently traversed. This keeps the boundary predictable and avoids invoking arbitrary sequence protocol methods.

Slot failures use zero-based integer locations. A bad second slot is located at [1]; a bad value nested inside that slot continues from [1, ...]. Field names remain available for introspection and documentation, but they are not Mapping keys or error-path segments for this positional boundary.

Defaults and arity¶

NamedTuple defaults follow Python's trailing-default rule:

class Record(NamedTuple):
    required: int
    optional: str = "default"


records = Contract(Record)
assert records.from_python([1]) == Record(1, "default")
assert records.from_python([1, "set"]) == Record(1, "set")

This contract accepts lengths one and two. An empty array reports missing at index 0; a third item reports unexpected at index 2. Nullable types affect the value in an occupied slot and do not make that position omittable.

Defaults are captured as canonical schema truth, validated when the Contract is created, and validated again when an omitted position uses them. This matters for the unusual but legal case of a mutable default whose contents change after compilation. Talea neither copies nor recomputes a static NamedTuple default.

JSON Schema uses ordered prefixItems, items: false, minItems equal to the required slot count, and maxItems equal to the total slot count. A safe non-sensitive, serializable trailing default is attached to its own prefixItems entry. prefixItems alone does not assert tuple length; the explicit minimum and maximum encode the same arity law used by conversion.

Metadata and composition¶

Field annotations use the normal Talea resolver. Constraints, Sensitive, descriptive schema metadata, and Representation therefore work inside a slot. A Representation can, for example, retain Decimal internally while accepting and emitting exact decimal text in the array.

Alias, including legacy names, is rejected because a positional slot has no external property name. ReadOnly and WriteOnly are also rejected: omitting an interior array position would create sparse directional semantics that Talea does not define. NamedTuple has no Talea serializer or validation hooks; enclosing Specs retain ownership of their own hooks.

NamedTuple fields compose beneath Specs, dataclasses, TypedDicts, sequences, mappings, ordinary unions, callable boundaries, Settings fields, incremental Contract input, and JSONL input. A branch in a tagged union may contain a NamedTuple, but a NamedTuple cannot itself become the object-property discriminator owner.

Nested output selection treats a NamedTuple as one leaf. Selecting the enclosing field is supported; descending by field name or index is not. The selection grammar remains canonical-name object selection and gains no index language.

Generics, recursion, and identity¶

Concrete generic specializations resolve each slot through the existing type parameter owner:

from typing import Generic, TypeVar

T = TypeVar("T")


class Pair(NamedTuple, Generic[T]):
    first: T
    second: T


assert Contract(Pair[int]).from_python([1, 2]) == Pair(1, 2)

An open Contract(Pair) is rejected rather than treating unresolved parameters as Any. PEP 695 aliases containing concrete NamedTuple types compose normally. Module-resolvable recursive and mutually recursive declarations reuse finite named references. A function-local forward-reference name that has already left scope remains unrecoverable; Talea does not inspect caller frames or add a namespace registry.

Two separately declared NamedTuple classes do not collapse merely because they have equal slot shapes. Declaration identity and generic arguments participate in the same named-schema identity owner used by other recursive structures.

Canonical schema and introspection¶

Annotation resolution creates one frozen NamedTupleSchema with the exact declared type, ordered NamedTupleField values, slot schemas, defaults, required count, and named identity. Strict validation, input, output, standards projection, recursion, resource accounting, and inspect_contract() all consume that node. Warm execution does not reread __annotations__, _fields, or _field_defaults.

from talea.introspection import inspect_contract
from talea.schema import NamedTupleSchema

info = inspect_contract(Contract(Record))
assert isinstance(info.schema, NamedTupleSchema)
assert tuple(field.name for field in info.schema.fields) == ("required", "optional")
assert info.schema.required_count == 1
assert info.schema.fields[1].default == "default"

The two schema-domain types are public from talea.schema, not the root package. There is no parallel rich NamedTuple introspection hierarchy, and the projection contains no generated functions, callbacks, locks, or caches.

Construction and trust¶

Supported declarations must retain Python's standard generated NamedTuple constructor shape. Talea validates or converts all positions first, then calls that constructor exactly once. A declaration with an incompatible mutated __new__ is rejected when its schema is resolved. Ordinary user methods are neither inspected nor executed by contract operations.

The outer tuple is immutable, but a slot may contain a mutable list, dictionary, dataclass, or Spec graph. Strict validation and output revalidate such current state instead of granting transitive trust. External traversal shares the existing ResourcePolicy depth, node, and error accounting; runtime cycles follow the normal cycle policy. Sensitive redacts Talea-owned failures, but a successful output operation is still application-authorized disclosure.

Talea never calls _asdict() and introduces no NamedTuple registry. Annotation evaluation remains trusted declaration-time Python work. Use the narrow list/tuple external boundary when positional records are part of the protocol; use an object-shaped owner when field-name evolution, aliases, sparse views, or property discriminators are requirements.

Complete market-data example¶

The executable example uses nested price levels, exact Decimal wire text, constraints, a trailing venue default, Sensitive data, a generic packet, recursion, a Spec envelope, standards projection, introspection, callable and incremental boundaries, JSONL arrays, and resource limits.

"""Compact market records with positional NamedTuple boundaries."""

from __future__ import annotations

from decimal import Decimal
from typing import Annotated, Generic, NamedTuple, TypeVar, cast

from talea import (
    Contract,
    Ge,
    Representation,
    ResourceLimitError,
    ResourcePolicy,
    Sensitive,
    Spec,
    ValidationError,
    validate_call,
)
from talea.introspection import inspect_contract
from talea.schema import NamedReferenceSchema, NamedTupleSchema, VariadicTupleSchema


def _load_decimal(value: str) -> Decimal:
    return Decimal(value)


def _dump_decimal(value: Decimal) -> str:
    return str(value)


type WireDecimal = Annotated[
    Decimal,
    Representation(input=str, load=_load_decimal, output=str, dump=_dump_decimal),
]


class PriceLevel(NamedTuple):
    """One price and its non-negative displayed quantity."""

    price: WireDecimal
    quantity: Annotated[int, Ge(0)]


class Quote(NamedTuple):
    """Compact positional quote used by the market-data protocol."""

    symbol: str
    bid: PriceLevel
    ask: PriceLevel
    venue: str = "XNAS"


class AuthenticatedQuote(NamedTuple):
    """Quote paired with a credential that Talea-owned failures redact."""

    quote: Quote
    token: Annotated[str, Sensitive()]


T = TypeVar("T")


class Packet(NamedTuple, Generic[T]):
    """Generic sequence-numbered market-data packet."""

    sequence: int
    payload: T


class Chain(NamedTuple):
    """Recursive positional record used for linked venue snapshots."""

    venue: str
    children: tuple[Chain, ...] = ()


class Snapshot(Spec):
    """Object-shaped envelope containing positional quote records."""

    source: str
    quote: Quote


quote_contract = Contract(Quote, policy=ResourcePolicy(max_depth=16, max_nodes=1_000))
quote = Quote(
    "AAPL",
    PriceLevel(Decimal("198.10"), 120),
    PriceLevel(Decimal("198.14"), 90),
)

assert quote_contract.validate(quote) is quote
try:
    quote_contract.validate(tuple(quote))  # type: ignore[arg-type]
except ValidationError as error:
    assert error.errors()[0]["location"] == []
else:
    raise AssertionError("strict validation accepted a plain tuple")

wire_quote: list[object] = ["AAPL", ["198.10", 120], ["198.14", 90]]
converted = quote_contract.from_python(wire_quote)
assert converted == quote
assert quote_contract.from_python(tuple(wire_quote)) == quote
assert quote_contract.from_json('["AAPL",["198.10",120],["198.14",90]]') == quote
assert converted.venue == "XNAS"
try:
    quote_contract.from_python({"symbol": "AAPL"})
except ValidationError as error:
    assert error.errors()[0]["code"] == "type"
else:
    raise AssertionError("a Mapping was accepted as a positional record")

try:
    quote_contract.from_python(["AAPL", ["198.10", -1], ["198.14", 90]])
except ValidationError as error:
    assert error.errors()[0]["location"] == [1, 1]
    assert error.errors()[0]["code"] == "greater_than_or_equal"
else:
    raise AssertionError("a negative displayed quantity was accepted")

invalid_secret = AuthenticatedQuote(quote, 7)  # ty: ignore[invalid-argument-type]
try:
    Contract(AuthenticatedQuote).validate(invalid_secret)
except ValidationError as error:
    assert error.errors()[0]["location"] == [1]
    assert error.errors()[0]["input"] == "<redacted>"
else:
    raise AssertionError("invalid sensitive state was accepted")

python_output = quote_contract.to_python(quote)
assert type(python_output) is tuple
assert python_output == ("AAPL", ("198.10", 120), ("198.14", 90), "XNAS")
assert quote_contract.to_json(quote) == '["AAPL",["198.10",120],["198.14",90],"XNAS"]'

snapshot = Snapshot.from_mapping({"source": "feed-a", "quote": wire_quote})
assert snapshot.quote is not quote
assert snapshot.quote == quote
packet_contract: Contract[Packet[Quote]] = Contract(Packet[Quote])
packet = packet_contract.from_python([42, wire_quote])
assert packet == Packet(42, quote)

chain_contract = Contract(Chain)
chain = chain_contract.from_python(["XNAS", (["BATS"],)])
assert chain == Chain("XNAS", (Chain("BATS"),))

schema = quote_contract.json_schema()
quote_schema = cast(dict[str, object], cast(dict[str, object], schema["$defs"])["Quote"])
assert quote_schema["type"] == "array"
assert quote_schema["minItems"] == 3
assert quote_schema["maxItems"] == 4
assert cast(list[object], quote_schema["prefixItems"])[3] == {
    "type": "string",
    "default": "XNAS",
}
openapi = quote_contract.openapi_schema()
openapi_schemas = cast(dict[str, object], cast(dict[str, object], openapi["components"])["schemas"])
openapi_quote = cast(dict[str, object], openapi_schemas["Quote"])
assert openapi_quote["type"] == "array"
assert openapi_quote["minItems"] == quote_schema["minItems"]
assert openapi_quote["maxItems"] == quote_schema["maxItems"]

info = inspect_contract(quote_contract)
assert isinstance(info.schema, NamedTupleSchema)
assert tuple(field.name for field in info.schema.fields) == ("symbol", "bid", "ask", "venue")
assert info.schema.required_count == 3
assert info.schema.fields[3].default == "XNAS"
try:
    info.schema.required_count = 4  # ty: ignore[invalid-assignment]
except (AttributeError, TypeError):
    pass
else:
    raise AssertionError("introspection schema was mutable")

recursive_schema = chain_contract._artifacts.schema
assert isinstance(recursive_schema, NamedTupleSchema)
children_schema = recursive_schema.fields[1].schema
assert isinstance(children_schema, VariadicTupleSchema)
assert isinstance(children_schema.item, NamedReferenceSchema)


@validate_call
def publish(value: Quote) -> Quote:
    """Publish an already-validated exact quote record."""

    return value


assert publish(quote) is quote
try:
    publish(tuple(quote))  # ty: ignore[invalid-argument-type]
except ValidationError:
    pass
else:
    raise AssertionError("callable validation accepted a plain tuple")

assert list(quote_contract.iter_python((wire_quote, tuple(wire_quote)))) == [quote, quote]
assert list(quote_contract.iter_jsonl(('["AAPL",["198.10",120],["198.14",90]]',))) == [quote]

try:
    quote_contract.from_python(wire_quote, policy=ResourcePolicy(max_nodes=3))
except ResourceLimitError as error:
    assert error.code == "nodes"
else:
    raise AssertionError("the positional graph exceeded max_nodes")