loom.rest.fastapi

FastAPI-specific REST layer for Loom.

Provides the FastAPI binding for compiled REST interfaces: - MsgspecJSONResponse — zero-copy JSON response. - bind_interfaces() — binds compiled routes to FastAPI. - create_fastapi_app() — composition root.

class loom.rest.fastapi.MsgspecJSONResponse(*args, **kwargs)[source]

Bases: Response

FastAPI Response subclass that encodes content with msgspec.json.

Drop-in replacement for fastapi.responses.JSONResponse with two advantages:

  • Native msgspec.Struct serialisation — no dict conversion needed.

  • Single encoding pass — content is written directly to bytes with no intermediate JSON string or Pydantic round-trip.

Parameters:

  • content – Any object supported by msgspec.json.encode.

  • status_code – HTTP status code. Defaults to 200.

  • headers – Additional response headers.

  • media_type – Defaults to "application/json".

  • background – Optional Starlette background task.

  • args (Any)

  • kwargs (Any)

Return type:

Any

Example:

return MsgspecJSONResponse(content=my_struct, status_code=201)
render(content)[source]

Encode content to JSON bytes using msgspec.json.encode.

Parameters:

content (object) – Object to serialise.

Returns:

UTF-8 encoded JSON bytes.

Return type:

bytes

loom.rest.fastapi.bind_interfaces(app, compiled_routes, factory, executor)[source]

Register compiled routes on a FastAPI application.

For each CompiledRoute, creates a dynamic async handler and registers it via app.add_api_route. Path parameters are inferred from the route path template and exposed in the handler signature so FastAPI validates and documents them correctly.

Nested JSON Schema $defs produced by msgspec/pydantic are collected into a shared component registry and returned. The caller is responsible for injecting these into components.schemas of the OpenAPI document.

Parameters:

  • app (fastapi.FastAPI) – FastAPI application instance to register routes on.

  • compiled_routes (Sequence[CompiledRoute]) – Ordered list of fully resolved routes produced by RestInterfaceCompiler.

  • factory (UseCaseFactory) – Use-case factory for constructing instances per request.

  • executor (RuntimeExecutor) – Runtime executor that drives the use-case pipeline.

Returns:

Mapping of schema name → JSON Schema fragment for all collected $defs that should appear under components.schemas.

Return type:

dict[str, Any]

Example:

compiler = RestInterfaceCompiler(use_case_compiler)
routes = compiler.compile(UserRestInterface)
component_schemas = bind_interfaces(app, routes, factory, executor)
loom.rest.fastapi.create_app(*config_paths, code_path=None, metrics_registry=None)[source]

Create a FastAPI application from one or more YAML config files.

Config files are merged left-to-right — later files override earlier ones. Each file may also declare a top-level includes list to pull in additional base files before its own values (see load_config()).

TraceIdMiddleware is mounted automatically when trace.enabled is true (the default). SQLAlchemy SQL echo inherits trace.enabled unless database.echo is set explicitly. Prometheus middleware is mounted when metrics.enabled is true.

Parameters:

  • *config_paths (str) – One or more paths to YAML configuration files.

  • code_path (str | None) – Optional override for app.code_path. Resolved relative to the first config file when not absolute.

  • metrics_registry (CollectorRegistry | None) – Optional Prometheus CollectorRegistry used for PrometheusMiddleware and the scrape endpoint. Defaults to the global registry. Pass a fresh CollectorRegistry() in tests to avoid ValueError: Duplicated timeseries when multiple apps with metrics.enabled: true are created in the same process.

Returns:

Configured fastapi.FastAPI application, ready to serve.

Return type:

FastAPI

Example — single config:

app = create_app("config/app.yaml")

Example — base + environment override:

app = create_app("config/base.yaml", "config/production.yaml")

Example — single file using inline includes:

# config/app.yaml
# includes:
#   - base.yaml
#   - secrets.yaml
app = create_app("config/app.yaml")
loom.rest.fastapi.create_fastapi_app(result, interfaces, *, middleware=(), defaults=None, **fastapi_kwargs)[source]

Create a FastAPI application from a bootstrap result and REST interfaces.

Compiles all RestInterface declarations via RestInterfaceCompiler, binds each compiled route to the FastAPI instance, and returns the ready application.

Compilation is fail-fast: any structural error (missing use-case plan, duplicate route, missing prefix) raises InterfaceCompilationError before the app starts accepting requests.

Parameters:

  • result (BootstrapResult) – Fully initialised BootstrapResult from bootstrap_app().

  • interfaces (Sequence[type[RestInterface[Any]]]) – RestInterface subclasses declaring which endpoints to expose. Compiled in declaration order.

  • middleware (Sequence[Any]) –

    ASGI middleware classes to register on the application. Added in declaration order (first = outermost wrapper). Accepts any class compatible with FastAPI.add_middleware. Example:

    from loom.rest.middleware import TraceIdMiddleware
from loom.prometheus import PrometheusMiddleware

app = create_fastapi_app(
    result,
    interfaces=[...],
    middleware=[TraceIdMiddleware, PrometheusMiddleware],
)

  • defaults (RestApiDefaults | None) – Global REST API defaults (pagination mode, profile policy). Falls back to RestApiDefaults when not provided.

  • **fastapi_kwargs (Any) – Additional keyword arguments forwarded to the FastAPI constructor (e.g. title, version, docs_url).

Returns:

Configured fastapi.FastAPI instance ready to serve requests.

Raises:

InterfaceCompilationError – If any interface fails structural validation.

Return type:

fastapi.FastAPI

Example:

app = create_fastapi_app(
    result,
    interfaces=[UserRestInterface, OrderRestInterface],
    defaults=RestApiDefaults(pagination_mode=PaginationMode.CURSOR),
    title="My API",
    version="2.0.0",
)