"""FastAPI application factory.
:func:`create_fastapi_app` is the composition root that wires together the
domain bootstrap result and REST interface declarations into a runnable
``FastAPI`` instance.
It is intentionally kept thin — all validation happens during
:class:`~loom.rest.compiler.RestInterfaceCompiler` compilation (fail-fast at
startup) and all request handling is delegated to
:func:`~loom.rest.fastapi.router_runtime.bind_interfaces`.
Usage::
result = bootstrap_app(
config=cfg,
use_cases=[CreateOrderUseCase, GetOrderUseCase],
modules=[register_repositories],
)
app = create_fastapi_app(
result,
interfaces=[OrderRestInterface],
observability_runtime=ObservabilityRuntime.noop(),
title="Orders API",
version="1.0.0",
)
"""
from __future__ import annotations
from collections.abc import Sequence
from typing import Any, cast
from fastapi import FastAPI
from loom.core.bootstrap.bootstrap import BootstrapResult
from loom.core.engine.executor import RuntimeExecutor
from loom.core.observability.runtime import ObservabilityRuntime
from loom.rest.compiler import RestInterfaceCompiler
from loom.rest.fastapi.router_runtime import bind_interfaces
from loom.rest.model import RestApiDefaults, RestInterface
# Type alias for ASGI middleware classes accepted by FastAPI.add_middleware.
_MiddlewareClass = Any
def _resolve_executor(result: BootstrapResult) -> RuntimeExecutor:
"""Return the application-scoped RuntimeExecutor registered at bootstrap."""
if not result.container.is_registered(RuntimeExecutor):
raise RuntimeError(
"RuntimeExecutor is not registered in the container. "
"Use bootstrap_app(...) or create_kernel(...) to build BootstrapResult."
)
return cast(RuntimeExecutor, result.container.resolve(RuntimeExecutor))
[docs]
def create_fastapi_app(
result: BootstrapResult,
interfaces: Sequence[type[RestInterface[Any]]],
*,
observability_runtime: ObservabilityRuntime | None = None,
middleware: Sequence[_MiddlewareClass] = (),
defaults: RestApiDefaults | None = None,
**fastapi_kwargs: Any,
) -> FastAPI:
"""Create a FastAPI application from a bootstrap result and REST interfaces.
Compiles all ``RestInterface`` declarations via
:class:`~loom.rest.compiler.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
:class:`~loom.rest.compiler.InterfaceCompilationError` before the app
starts accepting requests.
Args:
result: Fully initialised :class:`~loom.core.bootstrap.bootstrap.BootstrapResult`
from :func:`~loom.core.bootstrap.bootstrap.bootstrap_app`.
interfaces: ``RestInterface`` subclasses declaring which endpoints to
expose. Compiled in declaration order.
observability_runtime: Shared runtime used to emit lifecycle events
around each request.
middleware: 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=[...],
observability_runtime=ObservabilityRuntime.noop(),
middleware=[TraceIdMiddleware, PrometheusMiddleware],
)
defaults: Global REST API defaults (pagination mode, profile policy).
Falls back to :class:`~loom.rest.model.RestApiDefaults` when not
provided.
**fastapi_kwargs: Additional keyword arguments forwarded to the
``FastAPI`` constructor (e.g. ``title``, ``version``,
``docs_url``).
Returns:
Configured :class:`fastapi.FastAPI` instance ready to serve requests.
Raises:
InterfaceCompilationError: If any interface fails structural validation.
Example::
app = create_fastapi_app(
result,
interfaces=[UserRestInterface, OrderRestInterface],
defaults=RestApiDefaults(pagination_mode=PaginationMode.CURSOR),
observability_runtime=ObservabilityRuntime.noop(),
title="My API",
version="2.0.0",
)
"""
runtime = observability_runtime or ObservabilityRuntime.noop()
app = FastAPI(**fastapi_kwargs)
for mw_class in middleware:
app.add_middleware(mw_class)
interface_compiler = RestInterfaceCompiler(
result.compiler,
defaults=defaults,
)
executor = _resolve_executor(result)
all_routes = []
for iface in interfaces:
all_routes.extend(interface_compiler.compile(iface))
component_registry = bind_interfaces(
app,
all_routes,
result.factory,
executor,
observability_runtime=runtime,
)
if component_registry:
_register_openapi_components(app, component_registry)
return app
def _register_openapi_components(app: FastAPI, schemas: dict[str, Any]) -> None:
"""Patch ``app.openapi`` to inject ``schemas`` into ``components.schemas``.
Args:
app: FastAPI application whose OpenAPI generator to patch.
schemas: Mapping of component name → JSON Schema fragment collected
during route binding (nested ``$defs`` from msgspec/pydantic).
"""
original_openapi = app.openapi
def _openapi() -> dict[str, Any]:
doc = original_openapi()
doc.setdefault("components", {}).setdefault("schemas", {}).update(schemas)
return doc
app.openapi = _openapi # type: ignore[method-assign]