Tagged unions¶

Tagged unions make a payload's own required literal field select one branch. They are explicit: ordinary unions keep their existing validation and error semantics.

from typing import Annotated, Literal

from talea import Alias, Contract, Discriminator, Spec


class CardPayment(Spec):
    kind: Annotated[Literal["card"], Alias("type")]
    number: str


class BankTransfer(Spec):
    kind: Annotated[Literal["bank"], Alias("type")]
    iban: str


type Payment = Annotated[
    CardPayment | BankTransfer,
    Discriminator("type"),
]

payment = Contract(Payment).from_json(
    '{"type":"card","number":"4242"}'
)

Discriminator(name) may name the common Python field (kind) or its common external alias (type). The field declaration remains the only owner of both names. The tagged schema retains the resolved canonical name, external name, exact tags, JSON tag representations, sensitivity, and branch identities.

Branch contract¶

Every branch must be a Spec or every branch must be a TypedDict. A branch must carry the discriminator as a required key or field whose schema is one single-value Literal. Talea derives the tag from that Literal; there is no second Tag(...) declaration.

Supported tags are exact str, int, and bool values, plus Enum members whose JSON value is one of those types. Python tags remain type-sensitive, so True and 1 select different branches. Resolution also rejects tags that are distinct in Python but collapse to the same JSON representation, such as an IntEnum member with value 1 and the integer tag 1.

All branches must resolve to the same canonical field name and external name. Spec branch types must not be nominally overlapping. Concrete generic Spec specializations work normally; an open generic branch is not a concrete contract. Mixed Spec/TypedDict unions and unrelated hybrid alternatives are rejected. None is the one supported outer alternative:

type OptionalPayment = Annotated[
    CardPayment | BankTransfer | None,
    Discriminator("kind"),
]

Dispatch and execution¶

Strict validation of an existing Spec instance selects its branch by nominal identity and does not read a mapping tag. Mapping and JSON input locate the external discriminator, validate its exact type, select one branch, and run only that branch's compiled converter. TypedDict values necessarily dispatch from their key because dictionaries have no nominal branch identity.

Generated dispatch uses direct comparisons for two through four branches and a type-sensitive dictionary lookup bound directly to a compiled branch operation for five or more branches. Both strategies avoid validating rejected alternatives. Known-tag failures expose the selected branch's normal structural errors without a generic union wrapper.

Serialization selects a Spec branch by nominal identity or a TypedDict branch by its canonical tag. It then uses the selected branch's normal projection, so aliases, nested values, standard representations, and JSON encoding remain symmetric. A serialization hook on a discriminator field is rejected when the tagged contract resolves because it could contradict the round-trip contract. A hook also cannot replace an entire field whose reachable schema contains a tagged union; that would bypass branch projection and could silently emit a different or missing tag. Hooks on ordinary branch-body fields remain normal.

Errors¶

A missing key produces discriminator_missing; an exact, supported tag type with no branch produces discriminator_unknown. An unsupported tag type uses the existing type code. All three failures point at the discriminator path, including nested list and mapping segments.

Unknown-tag error projection includes discriminator and expected_tags as machine-readable fields. If any branch marks the discriminator Sensitive, tag failures redact the discriminator identity, received input, and expected tag set. A directional derivation of the containing Spec may omit the entire tagged field based on that field's metadata, but it does not rewrite branches.

TypedDict, recursion, and introspection¶

TypedDict branches use the same required single-Literal rule and preserve normal required/optional child keys, generic specialization, nested conversion, and detached output.

Recursive Spec graphs can contain tagged unions. Forward references finalize through the existing declaration graph and the tagged schema retains finite Spec references rather than copying complete branch declarations.

inspect_contract() and Spec field introspection expose TaggedUnionSchema. Its branch tuple is immutable and contains no public mutable dispatch table. json_schema() projects oneOf branches from this truth. openapi_schema() additionally emits a discriminator with the common external property name and a mapping to branch components. Recursive, generic, and TypedDict branches use the same finite definitions graph.

Production event stream¶

The executable example below uses four event types, UUID and datetime boundary representations, Decimal, aliases, a Sensitive authorization token, a generic EventEnvelope[T], strict JSON input/output, invalid tags, selected-branch failures, JSON Schema, and an OpenAPI discriminator map.

"""A tagged payment event boundary with direct dispatch and schema projection."""

from datetime import datetime
from decimal import Decimal
from typing import Annotated, Literal, cast
from uuid import UUID

from talea import Alias, Contract, Discriminator, Sensitive, Spec, ValidationError, WriteOnly


class PaymentAuthorized(Spec):
    kind: Annotated[Literal["payment.authorized"], Alias("type")]
    event_id: Annotated[UUID, Alias("eventId")]
    occurred_at: Annotated[datetime, Alias("occurredAt")]
    payment_id: Annotated[UUID, Alias("paymentId")]
    amount: Decimal
    authorization_token: Annotated[str, Alias("authorizationToken"), Sensitive(), WriteOnly()]


class PaymentDeclined(Spec):
    kind: Annotated[Literal["payment.declined"], Alias("type")]
    event_id: Annotated[UUID, Alias("eventId")]
    occurred_at: Annotated[datetime, Alias("occurredAt")]
    payment_id: Annotated[UUID, Alias("paymentId")]
    reason_code: Annotated[Literal["insufficient_funds", "issuer_declined"], Alias("reasonCode")]


class AccountFrozen(Spec):
    kind: Annotated[Literal["account.frozen"], Alias("type")]
    event_id: Annotated[UUID, Alias("eventId")]
    occurred_at: Annotated[datetime, Alias("occurredAt")]
    account_id: Annotated[UUID, Alias("accountId")]
    reason: str


class AccountUpdated(Spec):
    kind: Annotated[Literal["account.updated"], Alias("type")]
    event_id: Annotated[UUID, Alias("eventId")]
    occurred_at: Annotated[datetime, Alias("occurredAt")]
    account_id: Annotated[UUID, Alias("accountId")]
    changed_fields: Annotated[list[str], Alias("changedFields")]


type Event = Annotated[
    PaymentAuthorized | PaymentDeclined | AccountFrozen | AccountUpdated,
    Discriminator("type"),
]


class EventEnvelope[T](Spec):
    stream: str
    sequence: int
    payload: T


class EventActor(Spec):
    actor_id: Annotated[UUID, Alias("actorId")]
    display_name: Annotated[str, Alias("displayName")]
    internal_role: str


events: Contract[Event] = Contract(Event)
authorized_json = """{
  "type": "payment.authorized",
  "eventId": "10000000-0000-0000-0000-000000000001",
  "occurredAt": "2026-08-26T10:15:00Z",
  "paymentId": "20000000-0000-0000-0000-000000000002",
  "amount": "42.50",
  "authorizationToken": "gateway-secret"
}"""
event = events.from_json(authorized_json)
assert isinstance(event, PaymentAuthorized)
assert event.amount == Decimal("42.50")

projected = cast(dict[str, object], events.to_python(event))
assert projected["type"] == "payment.authorized"
assert projected["authorizationToken"] == "gateway-secret"
assert '"type":"payment.authorized"' in events.to_json(event)

envelope = EventEnvelope[Event](stream="payments", sequence=42, payload=event)
assert isinstance(envelope.payload, PaymentAuthorized)
assert EventEnvelope[Event].from_json(envelope.to_json()).sequence == 42


class DeliveryEnvelope(Spec):
    stream: str
    actor: EventActor
    payload: Event


delivery = DeliveryEnvelope(
    stream="payments",
    actor=EventActor(
        actor_id=UUID("30000000-0000-0000-0000-000000000003"),
        display_name="Payments gateway",
        internal_role="service-account",
    ),
    payload=event,
)
projected_delivery = delivery.to_dict(
    include={
        "stream": True,
        "actor": {"actor_id": True, "display_name": True},
        "payload": {
            "kind": True,
            "event_id": True,
            "payment_id": True,
            "amount": True,
            "reason_code": True,
            "account_id": True,
            "reason": True,
            "changed_fields": True,
        },
    }
)
assert projected_delivery == {
    "stream": "payments",
    "actor": {
        "actorId": UUID("30000000-0000-0000-0000-000000000003"),
        "displayName": "Payments gateway",
    },
    "payload": {
        "type": "payment.authorized",
        "eventId": UUID("10000000-0000-0000-0000-000000000001"),
        "paymentId": UUID("20000000-0000-0000-0000-000000000002"),
        "amount": Decimal("42.50"),
    },
}

try:
    events.from_json('{"type":"payment.refunded"}')
except ValidationError as error:
    detail = error.errors()[0]
    assert detail["code"] == "discriminator_unknown"
    assert detail["location"] == ["type"]
    # This union contains a Sensitive field, so discriminator diagnostics are
    # conservatively redacted with the rest of the boundary failure.
    assert detail["discriminator"] == "<redacted>"
else:
    raise AssertionError("an unknown event tag must fail before branch validation")

try:
    events.from_json(
        '{"type":"account.updated","eventId":"bad",'
        '"occurredAt":"2026-08-26T10:15:00Z","accountId":"bad","changedFields":[]}'
    )
except ValidationError as error:
    locations = {tuple(item["location"]) for item in error.errors()}
    assert ("eventId",) in locations
    assert ("accountId",) in locations
else:
    raise AssertionError("the selected branch must validate its nested fields")

schema = events.json_schema()
assert schema["$schema"] == "https://json-schema.org/draft/2020-12/schema"
assert "$defs" in schema
openapi = events.openapi_schema()
components = cast(dict[str, object], openapi["components"])
schemas = cast(dict[str, object], components["schemas"])
event_schema = cast(dict[str, object], schemas["Event"])
discriminator = cast(dict[str, object], event_schema["discriminator"])
mapping = cast(dict[str, str], discriminator["mapping"])
assert discriminator["propertyName"] == "type"
assert set(mapping) == {
    "payment.authorized",
    "payment.declined",
    "account.frozen",
    "account.updated",
}

The Sensitive token illustrates an important boundary rule: Sensitive redacts Talea-owned representation and failure snapshots, but successful serialization still follows the declared contract. If an event projection must omit a secret, declare a separate outward event type or explicitly exclude that field at the Spec boundary; do not treat security metadata as an output allow-list.

Tagged dispatch is a domain fit here because type is a protocol-level fact. An unknown event is not “the branch whose validation failed least”; it is an unsupported protocol message. Direct selection also gives framework tooling a canonical OpenAPI mapping and avoids creating diagnostics for branches the sender never selected.

Performance and when not to tag¶

Tagged dispatch runs only the selected branch. Small unions use direct comparisons; larger unions use a retained type-sensitive lookup. Cold resolution verifies tags and collisions once, while repeated Mapping/JSON input avoids the trial cost and branch-error allocation of an untagged union.

Do not add a discriminator merely to optimize a union whose data has no stable tag. The field becomes part of the external contract and must round-trip. An ordinary union remains clearer for small value alternatives such as int | str or when branch choice is genuinely structural. For recursive composition, see the tagged AST; for output shapes, see JSON Schema and OpenAPI.