Routes and server

Routes bind contracts to use cases. The server composes routes with concrete dependencies and ASGI concerns.

Bind a route

from tenchi.routes import route, route_group

from .contracts import create_todo_contract
from .use_cases.create_todo import create_todo


routes = route_group(
    route(create_todo_contract, create_todo),
)

route() checks the use-case signature immediately. A missing input, wrong parameter name, incompatible annotation, invalid header projector, or invalid presenter is a composition error.

Compose route groups

from tenchi.routes import route_group

from app.features.notes.routes import routes as note_routes
from app.features.todos.routes import routes as todo_routes


api_routes = route_group(
    todo_routes,
    note_routes,
    prefix="/api",
)

Groups flatten to an immutable route collection. errors= declares an application error across the group. Prefixes and shared errors create composition, not hidden runtime behavior.

Create the ASGI application

import os
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager

from starlette.applications import Starlette
from tenchi.server import create_app

from app.infra.port_wiring import ensure_schema, open_todo_repository
from app.server.context import AppContext
from app.server.routes import routes

DATABASE_PATH = os.environ.get("APP_DATABASE", "app.db")


def build_app(database_path: str = DATABASE_PATH) -> Starlette:
    @asynccontextmanager
    async def lifespan() -> AsyncIterator[str]:
        await ensure_schema(database_path)
        yield database_path

    @asynccontextmanager
    async def create_context(path: str) -> AsyncIterator[AppContext]:
        async with open_todo_repository(path) as todos:
            yield AppContext(todos=todos)

    return create_app(
        routes=routes,
        context_factory=create_context,
        lifespan=lifespan,
    )


app = build_app()

This matches the generated application's resource model. The lifespan prepares the database schema once at startup. Each request enters its own repository scope, which can commit on success, roll back on failure, and close before Tenchi finalizes the response. app.server.routes:routes composes the API, OpenAPI, Swagger UI, and health routes at the composition root.

create_app() returns a Starlette application and rejects duplicate or equivalent route templates and invalid context-factory shapes during composition. Operations on one path shape must use the same path-parameter names. Runtime matching is independent of declaration order: literal and constrained paths take precedence over broader parameterized paths. A GET contract handles HEAD automatically unless a matching explicit HEAD contract takes precedence.

Lifespan and request scopes

An async context manager can create process-scoped resources once:

from collections.abc import AsyncIterator
from contextlib import asynccontextmanager


@asynccontextmanager
async def lifespan() -> AsyncIterator[Database]:
    database = await Database.connect()
    try:
        yield database
    finally:
        await database.close()


def create_context(database: Database) -> AppContext:
    return AppContext(todos=SqlTodoRepository(database))


app = create_app(
    routes=routes,
    lifespan=lifespan,
    context_factory=create_context,
)

The context factory may return a context directly, await one, or return an async context manager. The final form creates request-scoped resources and guarantees cleanup before the response is finalized.

Hooks, middleware, and observers

• hooks= run after route matching and context creation, before input validation. They may return an enriched context and are the right boundary for authentication.
• webhooks= binds exact-body verifiers to contracts marked webhook=True. Verifiers run after ordinary hooks and body size/media checks, but before request parsing.
• middleware= accepts ordinary Starlette Middleware values for CORS, sessions, trusted hosts, compression, or app-specific ASGI behavior.
• observers= receive immutable RequestOutcome values after a matched request finalizes. Each outcome records its UTC completed_at before observer delivery. Observer failures are logged and cannot change the response.
• use_case_observers= receive immutable UseCaseOutcome values after the request scope closes, but only when the matched use case ran. The same observer contract works with execute(); completed_at records when the use-case call returned or raised, before later context cleanup.