Custom domain representations¶

Representation binds an application-owned Python type to explicit external input and output contracts at one Annotated position. Use it for values such as money, third-party identifiers, geographic coordinates, or immutable SDK types when the internal object itself is not Talea's boundary shape.

type MoneyValue = Annotated[
    Money,
    Representation(
        input=MoneyInput,
        load=load_money,
        output=MoneyOutput,
        dump=dump_money,
    ),
]

The resolved annotation owns one canonical RepresentationSchema: the strict internal schema, optional input schema, optional output schema, and private callback association. Strict validation, Python/JSON input, Python/JSON output, JSON Schema, OpenAPI, introspection, and nested selection all consume that same node. No callback registry or schema callback is involved.

Execution contracts¶

Input executes in this order:

1. validate the external value against input=;
2. call load exactly once;
3. validate its result against the internal annotation;
4. return the internal value.

Output executes in this order:

1. validate the current internal value under the normal containing contract;
2. call dump exactly once;
3. validate its result against output=;
4. apply the normal detached Python or JSON projector for that output schema;
5. for to_json(), encode the projected JSON tree once.

A wrong dump result raises SerializationError; it never escapes merely because a codec could encode it. Structured mutable callback results are detached by normal projection. Python and JSON output remain separate compiled operations, so standard Decimal, UUID, datetime, bytes, Enum, dataclass, Spec, TypedDict, and container rules remain authoritative.

One-way and bidirectional declarations¶

input and load form one complete pair; output and dump form another. Declare either pair or both:

• input-only supports strict validation and external input; output compilation raises SerializationError because no output direction exists;
• output-only supports strict validation and output; external input fails because no input direction exists;
• bidirectional supports all six Contract operations when every nested representation also declares the needed direction.

There is no fallback to repr, __dict__, identity, or an arbitrary object serializer for a missing direction.

Composition and nested selection¶

A reusable represented alias can appear under Specs, dataclasses, TypedDicts, lists, tuples, sets, frozensets, mapping values, unions, tagged branches, concrete generics, aliases, and recursive containing graphs. The callbacks stay attached to the represented position; containers do not consult a registry.

When output= is structural, nested include and exclude selection validates against that declared output schema and compiles a direct selected projector. Unknown output fields reject before dump executes. Talea validates the full dump result, then reads only selected output fields during projection; it does not serialize a complete dictionary and recursively post-filter it.

A field-local @serialize hook still overrides ordinary field output. Its return structure is undeclared and therefore remains an opaque selection leaf. Declared field-serializer output contracts are separate: @serialize(..., output=Summary) overrides one Spec field and projects the callback result through Summary. Representation remains the reusable annotation-position owner wherever a domain type occurs. Both reuse canonical Schema validation and output compilation, but neither declaration mechanism owns the other.

JSON Schema, OpenAPI, and introspection¶

json_schema(mode="input") and openapi_schema(mode="input") project input=. Output mode projects output=. A missing direction raises SchemaProjectionError. Schema tooling never executes load or dump, and callbacks cannot inject arbitrary schema fragments. PEP 695 alias identity continues to own reusable definition names; callback identity never appears in documents.

inspect_contract() and inspect_spec() expose callback-free frozen RepresentationInfo projections through their representations tuples. Each item contains the internal, input, and output canonical schemas plus has_loader and has_dumper. It exposes no callable, callable name, globals, generated source, lock, or cache.

Relationship to other features¶

Output constraints belong inside output=, for example output=Annotated[str, Pattern(...)]. Outer constraints still apply to the internal schema when meaningful. Metadata such as Title, Alias, and Sensitive retains its existing owner; Representation does not duplicate it.

Performance model¶

Representation adds work only at annotated positions. Warm generated input binds the loader and strict result validator directly; warm output binds the dumper, declared result validator, and mode-specific projector directly. It does not walk Schema metadata, acquire a lock, or consult a registry. Strict validation uses only the internal schema and never binds or calls the loader.

The permanent benchmark_representation task measures strict validation, scalar and structured Python/JSON input and output, containers, Spec, dataclass, TypedDict, generics, recursion, nested selection, failures, allocations, callback retention, generated call sites, and unrelated Contract(int)/ordinary-Spec canaries. Retain a Contract when it is reused; creating one inside every request turns schema resolution and compilation into hot-path work.

Trust, security, and round trips¶

Loaders and dumpers are trusted synchronous Python. Talea validates their results and safely transports documented failures, but cannot sandbox callback CPU, memory, I/O, mutation, reentrancy, or logging. ResourcePolicy governs external input traversal, not callback work or output size. A tiny internal value may deliberately dump a large structure.

Sensitive protects Talea-owned failure text, values, locations, and causes; it cannot stop a callback from logging a secret itself. Successful represented output is not automatically omitted. Use an explicit response shape and application authorization for disclosure policy.

Talea guarantees that a successful loader result satisfies the internal contract and a successful dumper result satisfies the output contract. It does not guarantee dump(load(x)) == x, load(dump(v)) == v, byte-for-byte reversibility, or preservation of noncanonical spelling. Canonicalization is allowed and is demonstrated by the money and identifier examples below.

Complete executable examples¶

The example covers a finance boundary, a ULID-like immutable identifier, nested containers, Spec/dataclass/TypedDict output, asymmetric schema modes, nested selection, one-way declarations, canonicalization, and Sensitive dump failure.

"""Executable custom-domain Representation examples."""

from dataclasses import dataclass
from decimal import Decimal
from enum import StrEnum
from typing import Annotated, TypedDict, cast

from talea import Contract, Representation, Sensitive, Spec
from talea.serialization import SerializationError


class Currency(StrEnum):
    """Currencies accepted by the payment boundary."""

    CHF = "CHF"
    EUR = "EUR"


class Money:
    """Application-owned financial value with normalized minor precision."""

    __slots__ = ("amount", "currency")

    def __init__(self, amount: Decimal, currency: Currency) -> None:
        self.amount = amount.quantize(Decimal("0.01"))
        self.currency = currency

    def __eq__(self, other: object) -> bool:
        return isinstance(other, Money) and (self.amount, self.currency) == (other.amount, other.currency)


class MoneyInput(TypedDict):
    """Accepted request representation."""

    amount: str
    currency: Currency


class MoneyOutput(TypedDict):
    """Canonical response representation."""

    amount: str
    currency: Currency


def load_money(value: MoneyInput) -> Money:
    """Construct the domain value from accepted boundary data."""

    return Money(Decimal(value["amount"]), value["currency"])


def dump_money(value: Money) -> MoneyOutput:
    """Produce the canonical structured response value."""

    return {"amount": str(value.amount), "currency": value.currency}


type MoneyValue = Annotated[
    Money,
    Representation(
        input=MoneyInput,
        load=load_money,
        output=MoneyOutput,
        dump=dump_money,
    ),
]


class Payment(Spec):
    """Payment with represented values at scalar and container positions."""

    order_id: str
    amount: MoneyValue
    fees: list[MoneyValue]


payment = Payment.from_json(
    '{"order_id":"ord-7","amount":{"amount":"12.500","currency":"CHF"},"fees":[{"amount":"0.20","currency":"CHF"}]}'
)
assert payment.amount == Money(Decimal("12.50"), Currency.CHF)
assert payment.to_dict() == {
    "order_id": "ord-7",
    "amount": {"amount": "12.50", "currency": Currency.CHF},
    "fees": [{"amount": "0.20", "currency": Currency.CHF}],
}
assert payment.to_json() == (
    '{"order_id":"ord-7","amount":{"amount":"12.50","currency":"CHF"},"fees":[{"amount":"0.20","currency":"CHF"}]}'
)
assert payment.to_dict(include={"amount": {"amount": True}}) == {"amount": {"amount": "12.50"}}


@dataclass(slots=True)
class LedgerLine:
    """Stdlib dataclass containing the same reusable contract."""

    amount: MoneyValue


class Settlement(TypedDict):
    """TypedDict containing the same reusable contract."""

    amount: MoneyValue


line = LedgerLine(Money(Decimal("3"), Currency.EUR))
assert Contract(LedgerLine).to_python(line) == {"amount": {"amount": "3.00", "currency": Currency.EUR}}
settlement: Settlement = {"amount": Money(Decimal("4"), Currency.CHF)}
assert Contract[Settlement](Settlement).to_json(settlement) == ('{"amount":{"amount":"4.00","currency":"CHF"}}')


class Ulid:
    """Small stand-in for an immutable third-party identifier type."""

    __slots__ = ("text",)

    def __init__(self, text: str) -> None:
        self.text = text.upper()

    def __eq__(self, other: object) -> bool:
        return isinstance(other, Ulid) and self.text == other.text


def load_ulid(value: str) -> Ulid:
    """Load a normalized identifier."""

    return Ulid(value)


def dump_ulid(value: Ulid) -> str:
    """Return canonical identifier text."""

    return value.text


type UlidValue = Annotated[
    Ulid,
    Representation(input=str, load=load_ulid, output=str, dump=dump_ulid),
]
ulids = Contract[list[Ulid]](list[UlidValue])
identifier_values = ulids.from_json('["01jabc"]')
assert identifier_values == [Ulid("01JABC")]
assert ulids.to_json(identifier_values) == '["01JABC"]'


schema_calls: list[str] = []


def counted_load(value: int) -> Money:
    schema_calls.append("load")
    return Money(Decimal(value), Currency.CHF)


def counted_dump(value: Money) -> MoneyOutput:
    schema_calls.append("dump")
    return dump_money(value)


type AsymmetricMoney = Annotated[
    Money,
    Representation(input=int, load=counted_load, output=MoneyOutput, dump=counted_dump),
]
asymmetric = Contract[Money](AsymmetricMoney)
input_schema = asymmetric.json_schema(mode="input")
output_schema = asymmetric.openapi_schema(mode="output")
input_definitions = cast(dict[str, object], input_schema["$defs"])
assert input_definitions["AsymmetricMoney"] == {"type": "integer"}
output_components = cast(dict[str, object], output_schema["components"])
output_definitions = cast(dict[str, object], output_components["schemas"])
assert "MoneyOutput" in output_definitions
assert schema_calls == []


type InputOnlyMoney = Annotated[
    Money,
    Representation(input=MoneyInput, load=load_money),
]
type OutputOnlyMoney = Annotated[
    Money,
    Representation(output=MoneyOutput, dump=dump_money),
]
assert Contract[Money](InputOnlyMoney).from_python({"amount": "1", "currency": Currency.CHF}) == Money(
    Decimal("1"), Currency.CHF
)
assert Contract[Money](OutputOnlyMoney).to_python(Money(Decimal("2"), Currency.EUR)) == {
    "amount": "2.00",
    "currency": Currency.EUR,
}
try:
    Contract[Money](InputOnlyMoney).to_python(Money(Decimal("1"), Currency.CHF))
except SerializationError as error:
    assert "no output direction" in str(error)
else:
    raise AssertionError("input-only Representation unexpectedly supported output")


class SecretToken:
    """Opaque application secret."""

    __slots__ = ("value",)

    def __init__(self, value: str) -> None:
        self.value = value


def reject_secret(value: SecretToken) -> str:
    """Demonstrate a hostile application-owned callback failure."""

    raise RuntimeError(f"unsafe callback detail: {value.value}")


type SecretValue = Annotated[
    SecretToken,
    Representation(output=str, dump=reject_secret),
    Sensitive(),
]
try:
    Contract[SecretToken](SecretValue).to_python(SecretToken("token-123"))
except SerializationError as error:
    assert error.__cause__ is None
    assert "token-123" not in str(error)
else:
    raise AssertionError("failing secret dumper unexpectedly succeeded")

Current limitations¶

Plain Contract(ArbitraryClass) remains unsupported unless that class has an ordinary Talea schema such as a dataclass. Representation declarations are explicit annotations; Talea provides no process-global discovery registry or generic representation factory. Callbacks are synchronous and trusted. There is no callback sandbox, output resource policy, custom format namespace, or user-defined error-code namespace. Arbitrary transforms still make input schema unknowable, and undeclared @serialize output remains structurally opaque.