import functools
import inspect
from collections.abc import Mapping
from contextlib import asynccontextmanager
from dataclasses import fields, is_dataclass
from typing import TYPE_CHECKING, Any, cast
import msgspec
from litestar.handlers.http_handlers.base import HTTPRouteHandler
from litestar.plugins import InitPlugin
from litestar.response import Response
if TYPE_CHECKING:
from collections.abc import AsyncGenerator
from litestar import Litestar, Request
from litestar.config.app import AppConfig
from litestar_vite.config import InertiaConfig
from litestar_vite.ipc import BaseIPCTransport, SSRCircuitBreaker
[docs]
class InertiaPlugin(InitPlugin):
"""Inertia plugin.
This plugin configures Litestar for Inertia.js support, including:
- Optional session-backed redirect state
- Exception handler for Inertia responses
- InertiaRequest and InertiaResponse as default classes
- Type encoders for StaticProp and DeferredProp
- Shared IPC transport and circuit breaker for SSR
Async Prop Resolution:
Async ``optional()``/``defer()``/``lazy()``/``once()`` callbacks are
pre-resolved by ``InertiaResponse`` on the request event loop before
the body is serialized. This guarantees they share the loop with
request-scoped async resources (asyncpg/aiosqlite/sqlspec sessions),
so callbacks can safely use those resources.
Example::
from litestar_vite.inertia import InertiaPlugin, InertiaConfig
app = Litestar(
plugins=[InertiaPlugin(InertiaConfig())],
)
"""
__slots__ = ("_circuit_breaker", "_ipc_transport", "config")
[docs]
def __init__(self, config: "InertiaConfig") -> None:
"""Initialize the plugin with Inertia configuration."""
self.config = config
self._ipc_transport: "BaseIPCTransport | None" = None
self._circuit_breaker: "SSRCircuitBreaker | None" = None
[docs]
@asynccontextmanager
async def lifespan(self, app: "Litestar") -> "AsyncGenerator[None, None]":
"""Lifespan to manage the shared SSR IPC transport and circuit breaker.
Args:
app: The :class:`Litestar <litestar.app.Litestar>` instance.
Yields:
An asynchronous context manager.
"""
owns_transport = False
ssr_config = self.config.ssr_config
if ssr_config is not None:
if ssr_config.circuit_breaker_enabled:
from litestar_vite.ipc import SSRCircuitBreaker
self._circuit_breaker = SSRCircuitBreaker(
failure_threshold=ssr_config.circuit_breaker_failure_threshold,
reset_timeout=ssr_config.circuit_breaker_reset_timeout,
)
from litestar_vite.plugin import VitePlugin
try:
vite_plugin: VitePlugin | None = app.plugins.get(VitePlugin)
except KeyError:
vite_plugin = None
is_dev_mode = vite_plugin.config.is_dev_mode if vite_plugin is not None else False
if not is_dev_mode:
from litestar_vite.ipc import StdioIPCTransport
if vite_plugin is not None:
self._ipc_transport = vite_plugin.get_ipc_transport(ssr_config=ssr_config)
else:
from litestar_vite.config import PathConfig
from litestar_vite.config._paths import resolve_ssr_bundle_path
self._ipc_transport = StdioIPCTransport(
command=ssr_config.command or ["node", str(resolve_ssr_bundle_path(PathConfig()))],
cwd=ssr_config.cwd,
)
owns_transport = True
try:
yield
finally:
if owns_transport and self._ipc_transport is not None:
await self._ipc_transport.close()
self._ipc_transport = None
self._circuit_breaker = None
@property
def circuit_breaker(self) -> "SSRCircuitBreaker | None":
"""Return the active SSRCircuitBreaker instance."""
return self._circuit_breaker
@property
def ipc_transport(self) -> "BaseIPCTransport | None":
"""Return the active BaseIPCTransport instance."""
return self._ipc_transport
[docs]
def on_app_init(self, app_config: "AppConfig") -> "AppConfig":
"""Configure application for use with Vite.
Args:
app_config: The :class:`AppConfig <litestar.config.app.AppConfig>` instance.
Returns:
The :class:`AppConfig <litestar.config.app.AppConfig>` instance.
"""
from litestar.exceptions import HTTPException, ValidationException
from litestar_vite.inertia.exception_handler import (
_register_exception_handlers, # pyright: ignore[reportPrivateUsage]
exception_to_http_response,
)
from litestar_vite.inertia.helpers import DeferredProp, StaticProp
from litestar_vite.inertia.middleware import InertiaMiddleware
from litestar_vite.inertia.precognition import create_precognition_exception_handler
from litestar_vite.inertia.request import InertiaRequest
from litestar_vite.inertia.response import InertiaBack, InertiaResponse
if app_config.response_class is InertiaResponse: # pyright: ignore[reportUnknownMemberType]
return app_config
exception_handlers: "dict[type[Exception] | int, Any]" = {
Exception: exception_to_http_response,
HTTPException: exception_to_http_response,
}
if self.config.precognition:
exception_handlers[ValidationException] = create_precognition_exception_handler(
fallback_handler=exception_to_http_response
)
_register_exception_handlers(exception_handlers=exception_handlers, default_handler=exception_to_http_response)
app_config.exception_handlers.update(exception_handlers) # pyright: ignore[reportUnknownMemberType]
app_config.request_class = InertiaRequest
app_config.response_class = InertiaResponse
app_config.middleware.append(InertiaMiddleware)
app_config.signature_types.extend([InertiaRequest, InertiaResponse, InertiaBack, StaticProp, DeferredProp])
app_config.type_encoders = {
StaticProp: lambda val: val.render(),
DeferredProp: lambda val: val.render(),
**(app_config.type_encoders or {}),
}
app_config.type_decoders = [
(lambda x: x is StaticProp, lambda t, v: t(v)),
(lambda x: x is DeferredProp, lambda t, v: t(v)),
*(app_config.type_decoders or []),
]
app_config.lifespan.append(self.lifespan) # pyright: ignore[reportUnknownMemberType]
app_config.on_startup.append( # pyright: ignore[reportUnknownMemberType]
functools.partial(_wrap_app_handlers, component_opt_keys=self.config.component_opt_keys)
)
return app_config
def _request_from_context(kwargs: "dict[str, Any]") -> "Request[Any, Any, Any] | None":
from litestar_vite.inertia.middleware import get_current_inertia_request
request = kwargs.get("request")
if request is not None:
return cast("Request[Any, Any, Any]", request)
return get_current_inertia_request()
def _as_inertia_prop_mapping(data: "Any") -> "Mapping[str, Any] | None":
"""Return a shallow field mapping for structured prop-bag return types.
A ``msgspec.Struct``, dataclass, or pydantic model returned from an Inertia
handler represents a bag of page props, mirroring a returned ``dict``. They
are converted to a shallow mapping (field name -> live value) so each field
becomes a top-level Inertia prop and special prop objects (``StaticProp`` /
``DeferredProp``) are preserved for the response prop pipeline. A deep
conversion is deliberately avoided so it matches ``dict`` handling exactly.
Args:
data: The handler return value.
Returns:
A shallow ``{field: value}`` mapping, or ``None`` when ``data`` is not a
recognized structured prop-bag type.
"""
obj: "Any" = data
model_fields: "Any" = getattr(cast("Any", type(obj)), "model_fields", None)
if model_fields is not None and hasattr(obj, "model_dump"):
return {name: getattr(obj, name) for name in model_fields}
if isinstance(obj, msgspec.Struct):
return {field.encode_name: getattr(obj, field.name) for field in msgspec.structs.fields(obj)}
if is_dataclass(obj) and not isinstance(obj, type):
return {field.name: getattr(obj, field.name) for field in fields(obj)}
return None
async def _resolve_inertia_response_data(data: "Any", request: "Request[Any, Any, Any]") -> "Any":
from litestar_vite.inertia.helpers import is_pagination_container
from litestar_vite.inertia.response import InertiaResponse
if isinstance(data, InertiaResponse):
await data.resolve_async_props(request)
return cast("InertiaResponse[Any]", data)
if isinstance(data, Response):
return cast("Response[Any]", data)
content: "Any"
if data is None or isinstance(data, Mapping) or is_pagination_container(data):
content = cast("Any", data)
else:
content = _as_inertia_prop_mapping(data)
if content is None:
return data
response: InertiaResponse[Any] = InertiaResponse(content=content)
response._defer_status_to_handler = True # pyright: ignore[reportPrivateUsage]
await response.resolve_async_props(request)
return response
def _wrap_handler_fn(handler: "HTTPRouteHandler") -> None:
"""Wrap ``handler.fn`` so async Inertia prop callbacks resolve inside the
DI ``AsyncExitStack`` frame.
Litestar resolves layered ``after_request`` hooks by keeping only
``after_request_handlers[-1]`` (see ``litestar/handlers/http_handlers/base.py``),
so a plugin-registered hook can be silently dropped by any user-defined
``after_request`` at router/controller/handler level. Wrapping ``fn`` directly
sidesteps that — the wrapper IS the handler, and runs inside the
``async with stack:`` block in ``_call_handler_function`` where
yield-based dependencies are still alive.
"""
if getattr(handler.fn, "_inertia_wrapped", False):
return
original = handler.fn
@functools.wraps(original) # pyright: ignore[reportUnknownArgumentType]
async def wrapped(**kwargs: "Any") -> "Any":
result = original(**kwargs)
if inspect.isawaitable(result):
result = await result
request = _request_from_context(kwargs)
if request is None:
return result
return await _resolve_inertia_response_data(result, request)
wrapped._inertia_wrapped = True # type: ignore[attr-defined] # pyright: ignore[reportFunctionMemberAccess]
handler._fn = wrapped # pyright: ignore[reportPrivateUsage]
handler.has_sync_callable = False
def _handler_supports_inertia(handler: "HTTPRouteHandler", *, component_opt_keys: "tuple[str, ...]") -> bool:
"""Return whether a route handler declares an Inertia page target."""
opt: dict[str, Any] = handler.opt or {}
return any(bool(opt.get(key)) for key in component_opt_keys)
def _wrap_app_handlers(app: "Litestar", *, component_opt_keys: "tuple[str, ...]" = ("component", "page")) -> None:
"""Idempotently wrap Inertia-capable HTTP route handlers on the live app.
Run after Litestar's full route registration so dynamically-attached
handlers (controllers instantiated late, plugins, etc.) are also wrapped.
"""
for route in app.routes:
for handler in getattr(route, "route_handlers", ()):
if isinstance(handler, HTTPRouteHandler) and _handler_supports_inertia(
handler, component_opt_keys=component_opt_keys
):
_wrap_handler_fn(handler)
