Litestar

Start here: Frameworks
Common recipes: micro.install(app) opens the app on startup and binds the active app inside route handlers, so patterns resolve their backends ambiently without explicit backend= wiring. Call it after Litestar(...), which builds the middleware stack at construction.

grelmicro.integrations.litestar

Litestar integration that opens a Grelmicro app and binds it per request.

install

install(
    app: Litestar, micro: Grelmicro, *, ambient: bool = True
) -> None

Wire micro into a Litestar app.

Opens async with micro: on startup and closes it after shutdown, so the components are registered before the first request. Startup hooks and lifespan managers already passed to Litestar(...) keep running.

Adds an after_exception hook that marks a request whose handler raised an unhandled exception. The idempotency middleware stores nothing for a marked request.

With Trace or Metrics registered, records the request span and the HTTP server metrics of every request. The request span, the access record and the security event of a request name the same route.

When ambient is True, wraps the app's ASGI handler so patterns resolve through Grelmicro.current() inside route handlers. The wrap sits outside every middleware Litestar built, so one that resolves a backend ambiently always runs inside the request scope.

Call it after the app is built, since Litestar builds its middleware stack at construction time:

from litestar import Litestar

from grelmicro import Grelmicro

micro = Grelmicro(uses=[...])
app = Litestar(route_handlers=[...])
micro.install(app)

Prefer the polymorphic micro.install(app), which detects the framework and calls this for you.

Warns MiddlewarePlacementWarning when an AuthenticatedRequestsMiddleware the app passed to Litestar(middleware=[...]) publishes metadata its router never routes.

PARAMETER	DESCRIPTION
`app`	The Litestar application to wire. TYPE: `Litestar`
`micro`	The `Grelmicro` app to open in the lifespan and bind per request. TYPE: `Grelmicro`
`ambient`	Wrap the app's ASGI handler with `GrelmicroMiddleware` so patterns resolve ambiently inside route handlers. Default `True`. Pass `False` to skip it. TYPE: `bool` DEFAULT: `True`

RAISES	DESCRIPTION
`TypeError`	If a flood limit would run behind the router, before anything is wired.

install_error_responses

install_error_responses(
    app: Litestar, errors: ErrorResponses
) -> None

Render grelmicro rejections in a standard format on a Litestar app.

Registers one exception handler per rejection grelmicro raises to turn a caller away, so a rate limiter, a bulkhead, an open circuit breaker, an elapsed deadline, or an idempotency conflict answers the client with an application/problem+json body instead of a 500.

Litestar looks a handler up through the raised exception's class hierarchy, so registering AdmissionError covers every rejection under it, including one a later release adds.

Call it after the app is built and before it serves, since Litestar resolves each route's handlers on its first request. micro.install(app) calls this when ErrorResponses() is registered, so a direct call is only for an app that never goes through install.

from grelmicro.http import ErrorResponses
from grelmicro.integrations.litestar import install_error_responses

install_error_responses(app, ErrorResponses())

install_middleware

install_middleware(
    app: Litestar, components: Sequence[Any]
) -> None

Add the ASGI middleware each registered component asks for.

A component that carries asgi_middleware() returns the middleware class and the arguments to build it with, and this wraps the app's ASGI handler with it. Registration order is wrapping order, so the first one registered is the outermost and answers first.

One that authenticates, AuthenticatedRequests, is wrapped outermost among ours whatever order it was registered in, so none of ours serves a request that was never authenticated.

Each one is wrapped inside the binding, so a middleware that resolves a backend ambiently finds the app bound.

Once one that answers is wrapped, every route renders its own exceptions, as Litestar does for a route that runs middleware. A raised HTTPException then reaches each of ours as the response the caller receives, so it is stored, replayed, and carries the fields ours add.

Call it after the app is built, since Litestar builds its middleware stack at construction time. micro.install(app) calls this with the components it found, so a direct call is only for an app that never goes through install.

PARAMETER	DESCRIPTION
`app`	The Litestar application to wire. TYPE: `Litestar`
`components`	The registered components that carry an ASGI middleware. TYPE: `Sequence[Any]`

RAISES	DESCRIPTION
`TypeError`	If a flood limit would run behind the router.

is_bound

is_bound(app: Litestar) -> bool

Return whether the per-request binding middleware is in place.

Called by Grelmicro.check_ambient_binding and Grelmicro.describe to catch an app that never had micro.install(app) called on it. True for both the wrap install adds and a DefineMiddleware(GrelmicroMiddleware) passed to Litestar(middleware=[...]).

PARAMETER	DESCRIPTION
`app`	The Litestar application to inspect. TYPE: `Litestar`

error_response

error_response(
    request: Request,
    *,
    status: int,
    detail: str | None = None,
    extensions: dict[str, Any] | None = None,
) -> Response

Answer from your own exception handler in the app's error format.

The Litestar counterpart of the Starlette helper. The format is read from the app, so a service that registered ErrorResponses.tmf() answers in TMF from here too.

from grelmicro.integrations.litestar import error_response


def handle(request: Request, exc: InsufficientFunds) -> Response:
    return error_response(
        request, status=409, detail="Not enough to cover this charge."
    )

PARAMETER	DESCRIPTION
`request`	The request being answered, which knows its app. TYPE: `Request`
`status`	HTTP status code of the response. TYPE: `int`
`detail`	Explanation of this occurrence, safe to show a client. TYPE: `str \| None` DEFAULT: `None`
`extensions`	Extra members to carry, where the format has room for them. TYPE: `dict[str, Any] \| None` DEFAULT: `None`

Authenticated

Authenticated(
    *, scopes: Sequence[str] = ()
) -> Callable[
    [ASGIConnection, BaseRouteHandler], Awaitable[None]
]

Require an authenticated caller holding every scope named.

A Litestar guard:

from litestar import delete

from grelmicro.integrations.litestar import Authenticated


@delete(
    "/orders/{order_id:int}",
    guards=[Authenticated(scopes=["orders:write"])],
)
async def cancel(order_id: int) -> None: ...

A caller with no credential is answered 401, and one lacking a scope 403, whose WWW-Authenticate challenge names the scopes. It needs a registered AuthenticatedRequests, which verifies the token before the handler runs.

Anonymous

Anonymous() -> dict[str, Any]

Serve this route without a credential.

Every route is authenticated once AuthenticatedRequests is registered. Pass this as the handler's opt on the one that is public:

from litestar import get

from grelmicro.integrations.litestar import Anonymous


@get("/catalog", opt=Anonymous())
async def catalog() -> list[Product]: ...

It is a mapping, so it merges with options of your own: opt={**Anonymous(), "tag": "public"}. micro.install(app) reads it off every handler, per method, so a public read keeps its writes authenticated, and the app is read again when it starts.

A token sent to it is verified: a valid one is the caller in request.user, and one that does not verify is answered 401.

It is {"exclude_from_auth": True}, the key Litestar's own authentication middleware reads, so a handler written for that one is public here too.

current_token

current_token(connection: ASGIConnection) -> VerifiedToken

Return the bearer token the request presented, once it verified.

from litestar import Request, post

from grelmicro.integrations.litestar import current_token


@post("/orders")
async def create_order(request: Request) -> dict[str, str]:
    token = current_token(request)
    response = await payments.post(
        "/charges", auth=payments_for_user.auth(token)
    )
    return response.json()

AuthenticatedRequests leaves it on every request it authenticates. Read it to act for the caller, such as exchanging it with a TokenExchange. Its repr never shows the token.

PARAMETER	DESCRIPTION
`app`	The Litestar application to wire. TYPE: `Litestar`
`errors`	The registered component that renders each rejection. TYPE: `ErrorResponses`

RAISES	DESCRIPTION
`TypeError`	If `scopes` is a single string.
`ValueError`	If a scope is not an OAuth scope token.