Production service boundary¶

A service boundary has three responsibilities: protect the process before and during conversion, turn external representations into a validated application value, and return failures without exposing secrets or implementation details. It should distinguish invalid data from exhausted resource budgets because the two failures have different operational meaning.

The example below is deliberately framework-neutral. handle_create() could sit inside a FastAPI, Lilya, Django, Starlette, Flask, queue-consumer, or RPC adapter without making Talea depend on any of those systems. It covers raw bytes, nested contracts, aliases, constraints, Sensitive credentials, a finite ResourcePolicy, structured validation errors, an application operation, a separate response contract, JSON output, and input/output OpenAPI fragments.

"""A framework-neutral account API from hostile bytes to a safe response."""

import json
from collections.abc import Callable
from typing import Annotated, Literal, cast
from uuid import UUID

from talea import (
    Alias,
    MaxLength,
    MinLength,
    ReadOnly,
    ResourceLimitError,
    ResourcePolicy,
    Sensitive,
    Spec,
    ValidationError,
    WriteOnly,
)


class Address(Spec):
    line_1: Annotated[str, Alias("line1"), MinLength(1), MaxLength(120)]
    city: Annotated[str, MinLength(1), MaxLength(80)]
    postcode: Annotated[str, MinLength(3), MaxLength(16)]
    country: Annotated[str, MinLength(2), MaxLength(2)]


class Credentials(Spec):
    password: Annotated[str, Sensitive(), WriteOnly(), MinLength(12), MaxLength(128)]


class UserCreate(Spec):
    email: Annotated[str, MinLength(3), MaxLength(254)]
    display_name: Annotated[str, Alias("displayName"), MinLength(1), MaxLength(80)]
    address: Address
    credentials: Credentials


class UserResponse(Spec):
    user_id: Annotated[UUID, Alias("id"), ReadOnly()]
    email: str
    display_name: Annotated[str, Alias("displayName")]
    address: Address
    status: Literal["active"] = "active"


class StoredUser(Spec):
    user_id: UUID
    email: str
    display_name: str
    address: Address


StoreUser = Callable[[UserCreate], StoredUser]


def create_user(request: UserCreate) -> StoredUser:
    """Stand in for application/domain work after boundary validation."""

    return StoredUser(
        user_id=UUID("12345678-1234-5678-1234-567812345678"),
        email=request.email,
        display_name=request.display_name,
        address=request.address,
    )


def error_response(error: ValidationError) -> str:
    return json.dumps({"errors": error.errors()}, separators=(",", ":"))


def handle_create(body: bytes, store: StoreUser = create_user) -> tuple[int, str]:
    """Translate transport and validation failures without choosing a framework."""

    try:
        request = UserCreate.from_json(
            body,
            policy=ResourcePolicy(
                max_input_bytes=4_096,
                max_depth=8,
                max_nodes=200,
                max_errors=10,
            ),
        )
    except ResourceLimitError as error:
        return 413, json.dumps({"error": error.code}, separators=(",", ":"))
    except ValidationError as error:
        return 422, error_response(error)

    stored = store(request)
    response = UserResponse(
        user_id=stored.user_id,
        email=stored.email,
        display_name=stored.display_name,
        address=stored.address,
    )
    return 201, response.to_json()


request_body = b"""{
  "email": "ada@example.test",
  "displayName": "Ada Lovelace",
  "address": {
    "line1": "1 Analytical Engine Way",
    "city": "London",
    "postcode": "SW1A 1AA",
    "country": "GB"
  },
  "credentials": {"password": "correct horse battery staple"}
}"""

status, body = handle_create(request_body)
assert status == 201
document = json.loads(body)
assert document["id"] == "12345678-1234-5678-1234-567812345678"
assert document["displayName"] == "Ada Lovelace"
assert "credentials" not in document

invalid = request_body.replace(b'"postcode": "SW1A 1AA"', b'"postcode": "X"')
status, body = handle_create(invalid)
assert status == 422
detail = json.loads(body)["errors"][0]
assert detail["location"] == ["address", "postcode"]
assert detail["code"] == "min_length"

oversized = b'{"email":"' + b"x" * 5_000 + b'"}'
status, body = handle_create(oversized)
assert status == 413
assert json.loads(body) == {"error": "input_size"}

input_schema = UserCreate.openapi_schema(mode="input")
output_schema = UserResponse.openapi_schema(mode="output")
input_root = cast(dict[str, object], input_schema["schema"])
output_root = cast(dict[str, object], output_schema["schema"])
assert cast(str, input_root["$ref"]).endswith("/UserCreate")
assert cast(str, output_root["$ref"]).endswith("/UserResponse")
input_components = cast(dict[str, object], input_schema["components"])
output_components = cast(dict[str, object], output_schema["components"])
input_definitions = cast(dict[str, object], input_components["schemas"])
output_definitions = cast(dict[str, object], output_components["schemas"])
assert "UserCreate" in input_definitions
assert "UserResponse" in output_definitions


class Permission(Spec):
    code: str
    source: str


class AccountProfile(Spec):
    display_name: Annotated[str, Alias("displayName")]
    address: Address
    permissions: list[Permission]
    internal_note: str


class AccountSnapshot(Spec):
    account_id: Annotated[UUID, Alias("id")]
    profile: AccountProfile
    revision: int


snapshot = AccountSnapshot(
    account_id=UUID("12345678-1234-5678-1234-567812345678"),
    profile=AccountProfile(
        display_name="Ada Lovelace",
        address=Address(line_1="1 Engine Way", city="London", postcode="SW1A 1AA", country="GB"),
        permissions=[Permission(code="account.read", source="role"), Permission(code="trade.read", source="grant")],
        internal_note="operator-only",
    ),
    revision=7,
)
public_snapshot = json.loads(
    snapshot.to_json(
        include={
            "account_id": True,
            "profile": {
                "display_name": True,
                "address": {"city": True, "country": True},
                "permissions": {"code": True},
            },
        }
    )
)
assert public_snapshot == {
    "id": "12345678-1234-5678-1234-567812345678",
    "profile": {
        "displayName": "Ada Lovelace",
        "address": {"city": "London", "country": "GB"},
        "permissions": [{"code": "account.read"}, {"code": "trade.read"}],
    },
}

Compose the 0.2 boundary capabilities¶

Talea 0.2 can keep a standard-library dataclass as a domain representation, validate and construct it through Contract, embed it in one canonical Spec API contract, derive explicit request/response/PATCH shapes, and project only the nested response fields an endpoint needs. No duplicate dataclass Spec, hand-maintained read/write field list, or recursive post-serialization filter is required.

The following executable flow uses canonical names for selection while aliases remain the emitted external names. The output-derived view cannot contain the write-only Sensitive password, the input-derived partial cannot contain the read-only request identifier, and apply_patch() preserves the source contract's complete-state validation.

"""Compose dataclass boundaries, directional views, PATCH, and nested output."""

import json
from dataclasses import dataclass
from typing import Annotated, Protocol, cast

from talea import (
    Alias,
    Contract,
    ReadOnly,
    ResourcePolicy,
    Sensitive,
    Spec,
    WriteOnly,
    apply_patch,
    derive_spec,
)


@dataclass(frozen=True, slots=True)
class PostalAddress:
    city: str
    country: str
    internal_note: str


@dataclass(frozen=True, slots=True)
class AccountRecord:
    record_id: Annotated[int, Alias("recordId")]
    address: PostalAddress
    permissions: list[str]


records = Contract(AccountRecord)
record = records.from_json(
    """{
      "recordId": 7,
      "address": {"city": "London", "country": "GB", "internal_note": "ops"},
      "permissions": ["account.read", "trade.read"]
    }""",
    policy=ResourcePolicy(max_depth=8, max_nodes=64),
)
assert type(record) is AccountRecord
assert records.validate(record) is record
record_mapping = cast(dict[str, object], records.to_python(record))
assert record_mapping["recordId"] == 7
assert json.loads(records.to_json(record))["address"]["city"] == "London"


class AccountBoundary(Spec):
    request_id: Annotated[int, Alias("requestId"), ReadOnly()]
    record: AccountRecord
    password: Annotated[str, Sensitive(), WriteOnly()]


AccountInput = derive_spec(AccountBoundary, mode="input", name="AccountInput")
AccountOutput = derive_spec(AccountBoundary, mode="output", name="AccountOutput")
AccountPatch = derive_spec(AccountBoundary, mode="input", partial=True, name="AccountPatch")


class AccountInputValue(Protocol):
    record: AccountRecord
    password: str


created = cast(
    AccountInputValue,
    AccountInput.from_json(
        """{
          "record": {
            "recordId": 7,
            "address": {"city": "London", "country": "GB", "internal_note": "ops"},
            "permissions": ["account.read", "trade.read"]
          },
          "password": "correct horse battery staple"
        }"""
    ),
)
assert type(created.record) is AccountRecord
assert "correct horse battery staple" not in repr(created)

source = AccountBoundary(
    request_id=42,
    record=created.record,
    password=created.password,
)
patch = AccountPatch.from_json('{"password":"replacement credential"}')
updated = apply_patch(source, patch)
assert updated.request_id == 42
assert updated.record is source.record
assert updated.record == record
assert updated.password == "replacement credential"

response = AccountOutput.from_mapping({"requestId": source.request_id, "record": records.to_python(source.record)})
projected = json.loads(
    response.to_json(
        include={
            "request_id": True,
            "record": {
                "record_id": True,
                "address": {"city": True, "country": True},
                "permissions": True,
            },
        }
    )
)
assert projected == {
    "requestId": 42,
    "record": {
        "recordId": 7,
        "address": {"city": "London", "country": "GB"},
        "permissions": ["account.read", "trade.read"],
    },
}

input_schema = json.dumps(AccountInput.json_schema(mode="input"), sort_keys=True)
output_schema = json.dumps(AccountOutput.json_schema(mode="output"), sort_keys=True)
assert '"requestId"' not in input_schema
assert '"password"' in input_schema
assert '"requestId"' in output_schema
assert '"password"' not in output_schema

Follow the ownership boundary¶

UserCreate.from_json() owns JSON decoding and contract conversion. Once it returns, create_user() receives a complete immutable request and can focus on domain behavior. It does not receive a partly populated model, and it does not need to rediscover whether displayName, UUID text, or a nested address was valid.

StoredUser is an application-facing value in this example. UserResponse is the external output contract. Keeping them separate prevents request-only credentials from accidentally appearing merely because one class was reused for every layer. Sensitive protects Talea-owned error/repr surfaces; it is not an instruction to remove a successfully validated field from serialization. Designing a response contract is therefore the primary allow-list.

The example maps malformed contract data to 422 and resource rejection to 413, but those numbers are illustrative application policy. Talea does not prescribe HTTP status codes, exception middleware, routes, authentication, or persistence.

Add PATCH without losing presence¶

For a partial update, derive the boundary once near the source declaration:

UserPatch = derive_spec(
    User,
    exclude=("user_id",),
    partial=True,
    name="UserPatch",
)

patch = UserPatch.from_json(raw_body, policy=request_policy)
changed = patch.present_fields
updated = apply_patch(existing_user, patch)

present_fields uses canonical Python names even when JSON uses aliases. An empty object has no changes; explicit None is a change only for an optional field; and explicitly sending the current default remains a change. Applying the patch creates a complete candidate and reruns whole-Spec invariants before returning it. The executable PATCH example covers empty, default-equal, nullable, aliased, sensitive, invalid, and whole-object cases.

Publish schemas without giving Talea route ownership¶

UserCreate.openapi_schema(mode="input") and UserResponse.openapi_schema(mode="output") return two keys: a root schema fragment and reusable components. A framework adapter can attach the root to its request or response description and merge the components into the framework-owned OpenAPI document. Talea does not generate paths or operations.

Use input and output modes deliberately. They can differ in requiredness and JSON representation, and callback-defined transforms or serializers can make one direction impossible to describe statically. See JSON Schema and OpenAPI for exact projection behavior.

Failure and operational guidance¶

Consume ValidationError.errors() and stable codes; do not parse rendered strings.
Log request identifiers and bounded error details, not raw bodies.
Treat ResourceLimitError.code, limit, and observed as operational data.
Set a stricter application policy when the endpoint shape is known. Do not disable every dimension merely because one valid document exceeded a limit.
Keep authentication, authorization, uniqueness, database state, and business rules in the application layer unless they are truly value invariants.

ResourcePolicy governs Talea-owned transport and compiled input work. It does not sandbox custom JSON codecs, Mapping implementations, callbacks, or regular expressions. Continue with the resource/security model, error handling, and troubleshooting.