"""Core Vite plugin implementation.
Houses :class:`VitePlugin` and its route-inspection helper. Kept as a private
module so ``litestar_vite.plugin`` stays a thin re-export surface.
"""
import importlib
import os
from contextlib import asynccontextmanager, contextmanager
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast
from urllib.parse import urlsplit
from litestar.exceptions import NotFoundException, SerializationException
from litestar.middleware import DefineMiddleware
from litestar.plugins import CLIPlugin, InitPlugin
from litestar.serialization import decode_json
from litestar.static_files import create_static_files_router # pyright: ignore[reportUnknownVariableType]
from litestar_vite.config import JINJA_INSTALLED, TRUE_VALUES, ExternalDevServer, TypeGenConfig
from litestar_vite.loader import ViteAssetLoader
from litestar_vite.plugin._process import ViteProcess
from litestar_vite.plugin._proxy import (
ProxyHeadersMiddleware,
ViteProxyMiddleware,
create_disabled_vite_hmr_handlers,
create_ssr_http_proxy_handler,
create_ssr_ws_proxy_handler,
create_vite_hmr_handler,
)
from litestar_vite.plugin._static import StaticPlacement, StaticServerConfig, StaticServerMount
from litestar_vite.plugin._utils import (
build_litestar_route_prefixes,
create_proxy_client,
is_non_serving_assets_cli,
is_non_serving_context,
log_fail,
log_warn,
pick_free_port,
set_app_environment,
set_environment,
static_not_found_handler,
vite_not_found_handler,
)
from litestar_vite.utils import read_hotfile_url
if TYPE_CHECKING:
from collections.abc import AsyncGenerator, Generator, Iterable
from typing import Literal
import httpx2
from click import Group
from litestar import Litestar
from litestar.config.app import AppConfig
from litestar.types import ControllerRouterHandler, ExceptionHandlersMap
from litestar_vite.config import InertiaSSRConfig, ViteConfig
from litestar_vite.fragments import FragmentEngine
from litestar_vite.handler import AppHandler
from litestar_vite.plugin._static import StaticFilesConfig
def _user_has_root_http_handler(route_handlers: "Iterable[ControllerRouterHandler]") -> bool:
"""Return True if any handler in ``route_handlers`` registers an HTTP route at ``/``.
Walks top-level entries (functions, Controllers, Routers). A match means the SSR
proxy must omit the bare ``/`` from its HTTP path list to avoid Litestar raising
``Handler already registered for path '/'`` at app construction.
"""
import inspect
from litestar import Controller, Router
from litestar.handlers.http_handlers import HTTPRouteHandler
from litestar.routes import HTTPRoute
def _is_root_path(prefix: str, path: str) -> bool:
joined = (prefix.rstrip("/") + "/" + path.lstrip("/")).rstrip("/")
return joined in {"", "/"}
def _walk(item: Any, prefix: str = "") -> bool:
if isinstance(item, HTTPRouteHandler):
return any(_is_root_path(prefix, p) for p in item.paths)
if isinstance(item, type) and issubclass(item, Controller):
ctrl_path_attr = getattr(item, "path", "")
ctrl_path = ctrl_path_attr if isinstance(ctrl_path_attr, str) else ""
new_prefix = prefix.rstrip("/") + "/" + ctrl_path.lstrip("/")
for _, member in inspect.getmembers(item):
if isinstance(member, HTTPRouteHandler) and any(_is_root_path(new_prefix, p) for p in member.paths):
return True
return False
if isinstance(item, Router):
full_root = (prefix.rstrip("/") or "/").rstrip("/") or "/"
for route in item.routes:
if isinstance(route, HTTPRoute):
full_path = (prefix.rstrip("/") + "/" + route.path.lstrip("/")).rstrip("/") or "/"
if full_path == full_root:
return True
return False
return False
return any(_walk(item) for item in route_handlers or [])
[docs]
class VitePlugin(InitPlugin, CLIPlugin):
"""Vite plugin for Litestar.
This plugin integrates Vite with Litestar, providing:
- Static file serving configuration
- Jinja2 template callables for asset tags
- Vite dev server process management
- Async asset loader initialization
Example::
from litestar import Litestar
from litestar_vite import VitePlugin, ViteConfig
app = Litestar(
plugins=[
VitePlugin(config=ViteConfig(dev_mode=True))
],
)
"""
__slots__ = (
"_asset_loader",
"_config",
"_fragment_engine",
"_ipc_dev_transports",
"_ipc_transport",
"_proxy_client",
"_proxy_target",
"_route_prefix_cache",
"_spa_handler",
"_static_files_config",
"_vite_process",
)
[docs]
def __init__(
self,
config: "ViteConfig | None" = None,
asset_loader: "ViteAssetLoader | None" = None,
static_files_config: "StaticFilesConfig | None" = None,
) -> None:
"""Initialize the Vite plugin.
Args:
config: Vite configuration. Defaults to ViteConfig() if not provided.
asset_loader: Optional pre-initialized asset loader.
static_files_config: Optional configuration for static file serving.
"""
from litestar_vite.config import ViteConfig
if config is None:
config = ViteConfig()
self._config = config
self._asset_loader = asset_loader
self._fragment_engine: "FragmentEngine | None" = None
self._ipc_transport: "Any | None" = None
self._ipc_dev_transports: dict[tuple[str, str, int], Any] = {}
self._vite_process: "ViteProcess | None" = None
self._static_files_config: "StaticFilesConfig | None" = static_files_config
self._proxy_client: "httpx2.AsyncClient | None" = None
self._proxy_target: "str | None" = None
self._route_prefix_cache: tuple[str, ...] | None = None
self._spa_handler: "AppHandler | None" = None
def _get_vite_process(self) -> ViteProcess:
"""Get or create the Vite process manager lazily."""
if self._vite_process is None:
self._vite_process = ViteProcess(executor=self._config.executor)
return self._vite_process
def _is_inert(self) -> bool:
"""Return whether HTTP-serving Vite wiring should be skipped."""
if self._config.enabled is False:
return True
if self._config.enabled is True:
return False
return is_non_serving_context()
[docs]
def get_ipc_transport(self, ssr_config: "InertiaSSRConfig | None" = None) -> Any:
"""Resolve the IPC transport for the active runtime mode.
In development mode, returns an HTTP/1.1 TCP transport targeting the
active Vite dev server's ``/__litestar_ssr__`` endpoint. In production
mode, returns a cached ``StdioIPCTransport`` targeting the built SSR worker.
Args:
ssr_config: Configuration supplied by a separately registered InertiaPlugin.
Returns:
Configured BaseIPCTransport instance.
"""
from litestar_vite.ipc import StdioIPCTransport, TCPStreamIPCTransport
if self._config.is_dev_mode:
host = self._config.host
port = self._config.port
scheme = self._config.protocol
hotfile_path = self._resolve_hotfile_path()
if hotfile_path.is_file():
hot_url = read_hotfile_url(hotfile_path)
if hot_url:
parsed = urlsplit(hot_url)
scheme = parsed.scheme or scheme
if parsed.hostname:
host = parsed.hostname
if parsed.port:
port = parsed.port
if host in {"::", "[::]", "0.0.0.0"} or (host == "localhost" and scheme == "http"): # noqa: S104
host = "127.0.0.1"
target = (scheme, host, port)
if target not in self._ipc_dev_transports:
self._ipc_dev_transports[target] = TCPStreamIPCTransport(
host=host, port=port, path="/__litestar_ssr__", scheme=scheme
)
return self._ipc_dev_transports[target]
if self._ipc_transport is None:
from litestar_vite.config._inertia import InertiaConfig
from litestar_vite.config._paths import resolve_ssr_bundle_path
inertia = self._config.inertia
ssr_config = ssr_config or (inertia.ssr_config if isinstance(inertia, InertiaConfig) else None)
command = ssr_config.command if ssr_config is not None else None
cwd = (ssr_config.cwd if ssr_config is not None else None) or self._config.root_dir
self._ipc_transport = StdioIPCTransport(
command=command or ["node", str(resolve_ssr_bundle_path(self._config.paths))], cwd=cwd
)
return self._ipc_transport
@property
def proxy_client(self) -> "httpx2.AsyncClient | None":
"""Return the pooled development HTTP proxy client, if initialized."""
return self._proxy_client
@property
def config(self) -> "ViteConfig":
"""Get the Vite configuration.
Returns:
The ViteConfig instance.
"""
return self._config
@property
def asset_loader(self) -> "ViteAssetLoader":
"""Get the asset loader instance.
Lazily initializes the loader if not already set.
Returns:
The ViteAssetLoader instance.
"""
if self._asset_loader is None:
self._asset_loader = ViteAssetLoader.initialize_loader(config=self._config)
return self._asset_loader
@property
def fragment_engine(self) -> "FragmentEngine":
"""Return the active FragmentEngine instance, initializing if needed."""
if self._fragment_engine is None:
from litestar_vite.fragments import FragmentEngine
self._fragment_engine = FragmentEngine(
config=self._config, asset_loader=self.asset_loader, transport_factory=self.get_ipc_transport
)
return self._fragment_engine
[docs]
async def render_fragment(
self, component: str, props: dict[str, Any] | None = None, mode: "Literal['static', 'island']" = "static"
) -> str:
"""Render a UI component fragment to HTML asynchronously.
Args:
component: Component path or name.
props: Optional properties dictionary passed to the component.
mode: Render mode, either 'static' (pure HTML) or 'island' (with client hydration).
Returns:
Rendered HTML string with prepended scoped CSS.
"""
return await self.fragment_engine.render_fragment(component, props, mode)
[docs]
def render_fragment_sync(
self, component: str, props: dict[str, Any] | None = None, mode: "Literal['static', 'island']" = "static"
) -> str:
"""Render a UI component fragment to HTML synchronously.
Useful for synchronous Jinja2 template rendering.
Args:
component: Component path or name.
props: Optional properties dictionary passed to the component.
mode: Render mode, either 'static' (pure HTML) or 'island' (with client hydration).
Returns:
Rendered HTML string with prepended scoped CSS.
"""
return self.fragment_engine.render_fragment_sync(component, props, mode)
@property
def spa_handler(self) -> "AppHandler | None":
"""Return the configured SPA handler when SPA mode is enabled.
Returns:
The AppHandler instance, or None when SPA mode is disabled/not configured.
"""
return self._spa_handler
[docs]
def get_static_server_config(self) -> StaticServerConfig:
"""Describe where the production static bundle should be served from.
Returns:
A ``StaticServerConfig`` whose ``placement`` is ``NATIVE`` with one mount when
the production bundle is eligible for direct web-server serving, or ``ASGI`` with
a diagnostic ``reason`` when Litestar's static router must serve.
"""
asset_url = self._config.asset_url
bundle_dir = self._resolve_bundle_dir()
reason = (
self._get_static_runtime_fallback_reason()
or self._get_static_semantics_fallback_reason(asset_url)
or self._get_static_bundle_fallback_reason(bundle_dir)
)
if reason is not None:
return StaticServerConfig(placement=StaticPlacement.ASGI, reason=reason)
return StaticServerConfig(
placement=StaticPlacement.NATIVE,
mounts=(StaticServerMount(route=asset_url.rstrip("/"), directory=bundle_dir.resolve()),),
)
def _get_static_runtime_fallback_reason(self) -> str | None:
"""Return why the active Vite runtime cannot use native static serving."""
if self._config.enabled is False:
return "Vite plugin is disabled."
if self._config.is_dev_mode:
return "Vite development mode requires Litestar's proxy and static routes."
if self._config.wants_html_proxy:
return "Vite framework/SSR routing requires Litestar's application handler."
if not self._config.set_static_folders:
return "Vite automatic static routing is disabled."
return None
def _get_static_semantics_fallback_reason(self, asset_url: str) -> str | None:
"""Return why Vite's Litestar routing semantics require ASGI serving."""
if self._static_files_config is not None and self._static_files_config.has_asgi_overrides():
overrides = ", ".join(self._static_files_config.asgi_override_fields())
return f"Vite static configuration contains ASGI-dependent options ({overrides}) that native serving cannot preserve."
user_opt = self._static_files_config.opt if self._static_files_config else None
if user_opt and not user_opt.get("exclude_from_auth", True):
return "Vite static assets explicitly opt into Litestar authentication."
parsed_asset_url = urlsplit(asset_url)
if (
not asset_url.startswith("/")
or asset_url.startswith("//")
or asset_url == "/"
or parsed_asset_url.scheme
or parsed_asset_url.netloc
or parsed_asset_url.query
or parsed_asset_url.fragment
):
return "Vite's asset URL must be a local absolute path below '/' for native serving."
return None
def _get_static_bundle_fallback_reason(self, bundle_dir: Path) -> str | None:
"""Return why the configured bundle is not a ready production build."""
try:
if not bundle_dir.is_dir():
return "Vite's production bundle directory is missing or empty."
if next(bundle_dir.iterdir(), None) is None:
return "Vite's production bundle directory is missing or empty."
except OSError:
return "Vite's production bundle directory cannot be read."
manifest_path = next((path for path in self._config.candidate_manifest_paths() if path.is_file()), None)
if manifest_path is None:
return (
None
if (bundle_dir / "index.html").is_file()
else ("Vite's production bundle has no manifest or built index.html marker.")
)
try:
manifest = decode_json(manifest_path.read_text(encoding="utf-8"))
except (OSError, UnicodeDecodeError, SerializationException):
return "Vite's production manifest cannot be read or parsed."
return None if isinstance(manifest, dict) else "Vite's production manifest must contain a JSON object."
def _resolve_bundle_dir(self) -> Path:
"""Resolve the bundle directory to an absolute path.
Returns:
The absolute path to the bundle directory.
"""
bundle_dir = Path(self._config.bundle_dir)
if not bundle_dir.is_absolute():
return self._config.root_dir / bundle_dir
return bundle_dir
def _resolve_hotfile_path(self) -> Path:
"""Resolve the path to the hotfile.
Returns:
The absolute path to the hotfile.
"""
return self._resolve_bundle_dir() / self._config.hot_file
def _write_hotfile(self, content: str) -> None:
"""Write content to the hotfile.
Args:
content: The content to write (usually the dev server URL).
"""
hotfile_path = self._resolve_hotfile_path()
hotfile_path.parent.mkdir(parents=True, exist_ok=True)
hotfile_path.write_text(content, encoding="utf-8")
def _resolve_dev_command(self) -> "list[str]":
"""Resolve the command to run for the dev server.
Returns:
The list of command arguments.
"""
ext = self._config.runtime.external_dev_server
if isinstance(ext, ExternalDevServer) and ext.enabled:
return ext.command or self._config.executor.start_command
if self._config.hot_reload:
return self._config.run_command
return self._config.build_watch_command
def _ensure_proxy_target(self) -> None:
"""Prepare proxy target URL and port for proxy modes (vite, proxy, ssr).
For all proxy modes in dev mode:
- Auto-selects a free port if VITE_PORT is not explicitly set
- Sets the port in runtime config for JS integrations to read
For 'vite' mode specifically:
- Forces loopback host unless VITE_ALLOW_REMOTE is set
- Sets _proxy_target directly (JS writes hotfile when server starts)
For 'proxy'/'ssr' modes:
- Port is written to .litestar.json for SSR framework to read
- SSR framework writes hotfile with actual URL when ready
- Proxy discovers target from hotfile at request time
"""
if not self._config.is_dev_mode:
return
if self._config.proxy_mode is None:
return
if os.getenv("VITE_PORT") is None and self._config.runtime.port == 5173:
self._config.runtime.port = pick_free_port()
if self._config.proxy_mode == "vite":
if self._proxy_target is not None:
return
if os.getenv("VITE_ALLOW_REMOTE", "False") not in TRUE_VALUES:
self._config.runtime.host = "127.0.0.1"
self._proxy_target = f"{self._config.protocol}://{self._config.host}:{self._config.port}"
def _configure_inertia(self, app_config: "AppConfig") -> "AppConfig":
"""Configure Inertia.js by registering an InertiaPlugin instance.
This is called automatically when `inertia` config is provided to ViteConfig.
Users can still use InertiaPlugin manually for more control.
Args:
app_config: The Litestar application configuration.
Returns:
The modified application configuration.
"""
from litestar_vite.inertia.plugin import InertiaPlugin
inertia_plugin = InertiaPlugin(config=self._config.inertia) # type: ignore[arg-type]
app_config = inertia_plugin.on_app_init(app_config)
if inertia_plugin not in app_config.plugins:
app_config.plugins.append(inertia_plugin)
return app_config
def _insert_dev_proxy_middleware(self, app_config: "AppConfig", middleware: "DefineMiddleware") -> None:
"""Insert a dev proxy middleware at the earliest safe position.
Ordering guarantees:
- Proxy headers middleware should run first when enabled, to ensure
scheme/host/client are normalized before proxy routing decisions.
- Vite proxy middleware should run before user middleware to avoid
unnecessary auth/session overhead for pure Vite/static requests.
"""
insert_at = 0
for idx, existing in enumerate(app_config.middleware):
if getattr(existing, "middleware", None) is ProxyHeadersMiddleware:
insert_at = idx + 1
break
app_config.middleware.insert(insert_at, middleware)
[docs]
def on_cli_init(self, cli: "Group") -> None:
"""Register CLI commands.
Args:
cli: The Click command group to add commands to.
"""
from litestar_vite.cli import vite_group
cli.add_command(vite_group)
def _configure_jinja_callables(self, app_config: "AppConfig") -> None:
"""Register Jinja2 template callables for Vite asset handling.
Args:
app_config: The Litestar application configuration.
"""
from litestar.plugins.jinja import JinjaTemplateEngine
from litestar_vite.fragments._jinja import vite_fragment, wrap_fragment_template_handlers
from litestar_vite.loader import render_asset_tag, render_hmr_client, render_routes, render_static_asset
template_config = app_config.template_config # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
if template_config and isinstance(
template_config.engine_instance, # pyright: ignore[reportUnknownMemberType]
JinjaTemplateEngine,
):
engine = template_config.engine_instance # pyright: ignore[reportUnknownMemberType]
engine.register_template_callable(key="vite_hmr", template_callable=render_hmr_client)
engine.register_template_callable(key="vite", template_callable=render_asset_tag)
engine.register_template_callable(key="vite_static", template_callable=render_static_asset)
engine.register_template_callable(key="vite_routes", template_callable=render_routes)
engine.register_template_callable(key="vite_fragment", template_callable=vite_fragment)
app_config.on_startup.append(wrap_fragment_template_handlers) # pyright: ignore[reportUnknownMemberType]
def _wants_jinja_callables(self, app_config: "AppConfig") -> bool:
"""Decide whether to register Jinja ``vite_*`` callables for this app.
The full gate requires three conditions, each of which is independently
load-bearing: the configured mode must be ``template`` (canonical), Jinja2
must be importable, and the user must have provided a ``TemplateConfig``
backed by ``JinjaTemplateEngine``. Returning ``False`` here is harmless:
callables are simply not registered and HTMX-without-Jinja, raw HTML, and
non-Jinja template engine consumers continue to work.
Args:
app_config: The Litestar application configuration.
Returns:
True when ``vite_hmr`` / ``vite`` / ``vite_static`` / ``vite_routes`` should be registered.
"""
if self._config.mode != "template" or not JINJA_INSTALLED:
return False
template_config = app_config.template_config # pyright: ignore[reportUnknownMemberType,reportUnknownVariableType]
if template_config is None:
return False
jinja_template_engine = importlib.import_module("litestar.plugins.jinja").JinjaTemplateEngine
return isinstance(template_config.engine_instance, jinja_template_engine) # pyright: ignore[reportUnknownMemberType]
def _configure_static_files(self, app_config: "AppConfig") -> None:
"""Configure static file serving for Vite assets.
The static files router serves real files (JS, CSS, images). SPA fallback (serving
index.html for client-side routes) is handled by the AppHandler.
Args:
app_config: The Litestar application configuration.
"""
bundle_dir = self._resolve_bundle_dir()
resource_dir = Path(self._config.resource_dir)
if not resource_dir.is_absolute():
resource_dir = self._config.root_dir / resource_dir
static_dir = Path(self._config.static_dir)
if not static_dir.is_absolute():
static_dir = self._config.root_dir / static_dir
static_dirs = [bundle_dir, resource_dir]
if static_dir.exists() and static_dir != bundle_dir:
static_dirs.append(static_dir)
opt: dict[str, Any] = {}
if self._config.exclude_static_from_auth:
opt["exclude_from_auth"] = True
user_opt = self._static_files_config.opt if self._static_files_config else None
if user_opt:
opt = {**opt, **user_opt}
opt["_vite_static_handler"] = True
base_config: dict[str, Any] = {
"directories": (static_dirs if self._config.is_dev_mode else [bundle_dir]),
"path": self._config.asset_url,
"name": "vite",
"html_mode": False,
"include_in_schema": False,
"exception_handlers": {NotFoundException: static_not_found_handler},
}
user_config = self._static_files_config.as_router_kwargs() if self._static_files_config else {}
static_files_config: dict[str, Any] = {**base_config, **user_config}
router = create_static_files_router(**static_files_config)
for route in router.routes:
for handler in getattr(route, "route_handlers", []):
handler.opt.update(opt)
app_config.route_handlers.append(router)
def _configure_dev_proxy(self, app_config: "AppConfig") -> None:
"""Configure dev proxy middleware and handlers based on the canonical mode.
Args:
app_config: The Litestar application configuration.
"""
hotfile_path = self._resolve_hotfile_path()
if self._config.wants_html_proxy:
self._configure_ssr_proxy(app_config, hotfile_path)
else:
self._configure_vite_proxy(app_config, hotfile_path)
def _resolve_hmr_path(self) -> "str | None":
"""Return the local HMR websocket path for the configured asset URL."""
asset_url = self._config.asset_url
if not asset_url.startswith("/"):
return None
return f"{asset_url.rstrip('/')}/vite-hmr"
def _configure_vite_proxy(self, app_config: "AppConfig", hotfile_path: Path) -> None:
"""Configure Vite proxy mode (allow list).
Args:
app_config: The Litestar application configuration.
hotfile_path: Path to the hotfile.
"""
self._ensure_proxy_target()
self._insert_dev_proxy_middleware(
app_config,
DefineMiddleware(
ViteProxyMiddleware,
hotfile_path=hotfile_path,
asset_url=self._config.asset_url,
resource_dir=self._config.resource_dir,
bundle_dir=self._config.bundle_dir,
root_dir=self._config.root_dir,
http2=self._config.http2,
plugin=self,
),
)
hmr_path = self._resolve_hmr_path()
if hmr_path is not None:
app_config.route_handlers.append(
create_vite_hmr_handler(hotfile_path=hotfile_path, hmr_path=hmr_path, asset_url=self._config.asset_url)
)
def _configure_disabled_hmr_handler(self, app_config: "AppConfig") -> None:
"""Register a clean-close websocket sink for stale Vite HMR clients."""
hmr_path = self._resolve_hmr_path()
if hmr_path is not None:
app_config.route_handlers.extend(create_disabled_vite_hmr_handlers(hmr_path=hmr_path))
def _configure_ssr_proxy(self, app_config: "AppConfig", hotfile_path: Path) -> None:
"""Configure SSR proxy mode for framework dev servers.
Registers an HTTP catch-all route handler plus a WebSocket HMR handler at
``/`` and ``/{path:path}``. Litestar dispatches matched routes directly, so
any path the user has not registered falls through to the framework's dev
server. Litestar's per-route middleware never runs on unmatched paths, so a
true route is required (see commit history: an earlier middleware-based
approach hit 405 Method Not Allowed because the WS handler at ``/`` was
matched first).
Collision avoidance: when a user has already registered an HTTP handler at
``/``, the plugin omits ``/`` from the HTTP catch-all's path list (keeping
``/{path:path}``) so Litestar does not raise
``Handler already registered for path '/'`` at app construction.
``ViteProxyMiddleware`` is also registered for asset URLs since framework
dev servers commonly ship Vite-bundled assets.
Args:
app_config: The Litestar application configuration.
hotfile_path: Path to the hotfile.
"""
self._ensure_proxy_target()
external = self._config.external_dev_server
static_target = external.target if external else None
ssr_hotfile = hotfile_path if static_target is None else None
user_owns_root = _user_has_root_http_handler(app_config.route_handlers)
http_paths = ["/{path:path}"] if user_owns_root else ["/", "/{path:path}"]
app_config.route_handlers.append(
create_ssr_http_proxy_handler(
target=static_target,
hotfile_path=ssr_hotfile,
http2=external.http2 if external else True,
plugin=self,
paths=http_paths,
)
)
app_config.route_handlers.append(create_ssr_ws_proxy_handler(target=static_target, hotfile_path=ssr_hotfile))
self._insert_dev_proxy_middleware(
app_config,
DefineMiddleware(
ViteProxyMiddleware,
hotfile_path=hotfile_path,
asset_url=self._config.asset_url,
resource_dir=self._config.resource_dir,
bundle_dir=self._config.bundle_dir,
root_dir=self._config.root_dir,
http2=self._config.http2,
plugin=self,
),
)
hmr_path = self._resolve_hmr_path()
if hmr_path is not None:
app_config.route_handlers.append(
create_vite_hmr_handler(hotfile_path=hotfile_path, hmr_path=hmr_path, asset_url=self._config.asset_url)
)
[docs]
def on_app_init(self, app_config: "AppConfig") -> "AppConfig":
"""Configure the Litestar application for Vite.
This method wires up supporting configuration for dev/prod operation:
- Adds types used by generated handlers to the signature namespace.
- Ensures a consistent NotFound handler for asset/proxy lookups.
- Registers optional Inertia and Jinja integrations.
- Configures static file routing when enabled.
- Configures dev proxy middleware based on proxy_mode.
- Creates/initializes the SPA handler where applicable and registers lifespans.
Args:
app_config: The Litestar application configuration.
Returns:
The modified application configuration.
"""
from litestar import Response
from litestar.connection import Request as LitestarRequest
from litestar_vite.handler import AppHandler
if self._is_inert():
return app_config
app_config.signature_namespace["Response"] = Response
app_config.signature_namespace["Request"] = LitestarRequest
if self._config.trusted_proxies is not None:
app_config.middleware.insert(
0, DefineMiddleware(ProxyHeadersMiddleware, trusted_hosts=self._config.trusted_proxies)
)
handlers: ExceptionHandlersMap = cast("ExceptionHandlersMap", app_config.exception_handlers or {}) # pyright: ignore
if NotFoundException not in handlers:
handlers[NotFoundException] = vite_not_found_handler
app_config.exception_handlers = handlers # pyright: ignore[reportUnknownMemberType]
if self._config.inertia is not None:
app_config = self._configure_inertia(app_config)
if self._wants_jinja_callables(app_config):
self._configure_jinja_callables(app_config)
skip_static = (
self._config.wants_html_proxy
and self._config.is_dev_mode
and self._config.runtime.external_dev_server is not None
)
if self._config.set_static_folders and not skip_static:
self._configure_static_files(app_config)
dev_proxy_configured = (
self._config.is_dev_mode and self._config.proxy_mode is not None and not is_non_serving_assets_cli()
)
if dev_proxy_configured:
self._configure_dev_proxy(app_config)
elif self._config.set_static_folders:
self._configure_disabled_hmr_handler(app_config)
use_spa_handler = self._config.spa_handler and (
self._config.registers_html_catchall or self._config.wants_html_proxy
)
use_spa_handler = use_spa_handler or (
self._config.wants_html_proxy
and not self._config.is_dev_mode
and self._config.runtime.external_dev_server is not None
)
if use_spa_handler:
self._spa_handler = AppHandler(self._config, csrf_config=app_config.csrf_config)
app_config.route_handlers.append(self._spa_handler.create_route_handler())
elif self._config.mode == "hybrid":
self._spa_handler = AppHandler(self._config, csrf_config=app_config.csrf_config)
app_config.lifespan.append(self.lifespan) # pyright: ignore[reportUnknownMemberType]
return app_config
[docs]
def get_route_prefixes(self, app: "Litestar") -> tuple[str, ...]:
"""Return the immutable route-prefix cache owned by this plugin."""
if self._route_prefix_cache is None:
self._route_prefix_cache = build_litestar_route_prefixes(app, self._config.runtime.extra_route_prefixes)
return self._route_prefix_cache
def _check_health(self) -> None:
"""Check if the Vite dev server is running and ready.
Polls the dev server URL for up to 5 seconds.
"""
import time
import urllib.request
url = f"{self._config.protocol}://{self._config.host}:{self._config.port}/__vite_ping"
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
for _ in range(50):
try:
with opener.open(url, timeout=0.1):
return
except OSError:
time.sleep(0.1)
log_fail("Vite dev server did not become ready.")
def _run_health_check(self) -> None:
"""Run the appropriate health check based on proxy mode."""
match self._config.proxy_mode:
case "proxy":
self._check_ssr_health(self._resolve_hotfile_path())
case _:
self._check_health()
def _check_ssr_health(self, hotfile_path: Path, timeout: float = 10.0) -> bool:
"""Wait for SSR framework to be ready via hotfile.
Polls intelligently for the hotfile and validates HTTP connectivity.
Exits early as soon as the server is confirmed ready.
Args:
hotfile_path: Path to the hotfile written by the SSR framework.
timeout: Maximum time to wait in seconds (default 10s).
Returns:
True if SSR server is ready, False if timeout reached.
"""
import time
import urllib.error
import urllib.request
start = time.time()
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
while time.time() - start < timeout:
if hotfile_path.exists():
try:
url = read_hotfile_url(hotfile_path)
if url:
with opener.open(url, timeout=0.5) as resp:
if getattr(resp, "status", 200) < 500:
return True
except urllib.error.HTTPError as exc:
if exc.code < 500:
return True
except OSError:
pass
time.sleep(0.1)
log_fail(f"SSR framework server did not become ready within {timeout}s.")
return False
def _export_types_sync(self, app: "Litestar") -> None:
"""Export type metadata synchronously on startup.
This exports OpenAPI schema, route metadata (JSON), typed routes (TypeScript),
and Inertia pages metadata when type generation is enabled. The Vite plugin
watches these files and triggers @hey-api/openapi-ts when they change.
Uses the shared `export_integration_assets` function to guarantee
byte-identical output between CLI and plugin.
Args:
app: The Litestar application instance.
"""
from litestar_vite.codegen import export_integration_assets, typegen_outputs_requested
if not isinstance(self._config.types, TypeGenConfig):
return
if not typegen_outputs_requested(self._config.types):
return
try:
export_integration_assets(app, self._config)
except (OSError, TypeError, ValueError, ImportError): # pragma: no cover
log_warn(
"Vite type metadata export failed; run 'litestar assets generate-types' for details.",
level=self._config.logging_config.level,
)
[docs]
@contextmanager
def server_lifespan(self, app: "Litestar") -> "Generator[None, None, None]":
"""Server-level lifespan context manager (runs ONCE per server, before workers).
This is called by Litestar CLI before workers start. It handles:
- Environment variable setup (with logging)
- Vite dev server process start/stop (ONE instance for all workers)
- Type export on startup
Note: SPA handler and asset loader initialization happens in the per-worker
`lifespan` method, which is auto-registered in `on_app_init`.
Hotfile behavior: the JS plugin's listening callback writes the authoritative target once
the Vite socket is ready. Python pre-writes only user-asserted external targets whose dev
servers do not run the JS plugin.
Args:
app: The Litestar application instance.
Yields:
None
"""
if self._is_inert():
yield
return
if self._config.is_dev_mode:
self._ensure_proxy_target()
if self._config.set_environment:
set_environment(config=self._config, app=app)
set_app_environment(app)
self._export_types_sync(app)
if self._config.is_dev_mode and self._config.runtime.start_dev_server:
ext = self._config.runtime.external_dev_server
is_external = isinstance(ext, ExternalDevServer) and ext.enabled
command_to_run = self._resolve_dev_command()
if is_external and isinstance(ext, ExternalDevServer) and ext.target:
self._write_hotfile(ext.target)
vite_process: ViteProcess | None = None
try:
vite_process = self._get_vite_process()
vite_process.start(command_to_run, self._config.root_dir)
if self._config.health_check and not is_external:
self._run_health_check()
yield
finally:
if vite_process is not None:
vite_process.stop()
else:
yield
[docs]
@asynccontextmanager
async def lifespan(self, app: "Litestar") -> "AsyncGenerator[None, None]":
"""Worker-level lifespan context manager (runs per worker process).
This is auto-registered in `on_app_init` and handles per-worker initialization:
- Environment variable setup (silently - each worker needs process-local env vars)
- Asset loader initialization
- SPA handler initialization
- Route metadata injection
Note: The Vite dev server process is started in `server_lifespan`, which
runs ONCE per server before workers start.
Args:
app: The Litestar application instance.
Yields:
None
"""
from litestar_vite.loader import ViteAssetLoader
if self._config.set_environment:
set_environment(config=self._config, app=app)
set_app_environment(app)
if self._asset_loader is None:
self._asset_loader = ViteAssetLoader(config=self._config)
await self._asset_loader.initialize()
if self._spa_handler is not None and not self._spa_handler.is_initialized:
await self._spa_handler.initialize_async(vite_url=self._proxy_target, manifest=self._asset_loader.manifest)
is_ssr_mode = self._config.wants_html_proxy
if not self._config.is_dev_mode and not self._config.has_built_assets() and not is_ssr_mode:
log_warn(
"Vite dev server is disabled (dev_mode=False) but no index.html was found. "
"Run your front-end build or set VITE_DEV_MODE=1 to enable HMR.",
level=self._config.logging_config.level,
)
if self._config.is_dev_mode and self._config.proxy_mode is not None:
self._proxy_client = create_proxy_client(http2=self._config.http2)
try:
yield
finally:
if self._proxy_client is not None:
await self._proxy_client.aclose()
self._proxy_client = None
if self._spa_handler is not None:
await self._spa_handler.shutdown_async()
if self._fragment_engine is not None:
await self._fragment_engine.close()
for transport in self._ipc_dev_transports.values():
await transport.close()
self._ipc_dev_transports.clear()
if self._ipc_transport is not None:
await self._ipc_transport.close()
self._ipc_transport = None
