"""SPA mode handler for Vite integration.
This module provides :class:`~litestar_vite.handler.AppHandler` which manages serving
the Single Page Application (SPA) HTML in both development and production modes.
In dev mode, it proxies requests to the Vite dev server for HMR support.
In production, it serves the built index.html with async caching.
"""
import logging
from contextlib import suppress
from pathlib import Path
from typing import TYPE_CHECKING, Any, NoReturn
import anyio
import httpx
from litestar import get
from litestar.exceptions import ImproperlyConfiguredException, SerializationException
from litestar.serialization import decode_json, encode_json
from litestar_vite._handler.routing import spa_handler_dev, spa_handler_prod
from litestar_vite.html_transform import inject_head_script, set_data_attribute, transform_asset_urls
if TYPE_CHECKING:
from litestar.connection import Request
from litestar.types import Guard # pyright: ignore[reportUnknownVariableType]
from litestar_vite.config import SPAConfig, ViteConfig
logger = logging.getLogger("litestar_vite")
[docs]
class AppHandler:
"""Handler for serving SPA HTML in both dev and production modes."""
__slots__ = (
"_cached_bytes",
"_cached_html",
"_cached_transformed_html",
"_config",
"_http_client",
"_http_client_sync",
"_initialized",
"_manifest",
"_spa_config",
"_vite_url",
)
[docs]
def __init__(self, config: "ViteConfig") -> None:
"""Initialize the SPA handler.
Args:
config: The Vite configuration.
"""
self._config = config
self._spa_config: "SPAConfig | None" = config.spa_config
self._cached_html: "str | None" = None
self._cached_bytes: "bytes | None" = None
self._cached_transformed_html: "str | None" = None
self._initialized = False
self._http_client: "httpx.AsyncClient | None" = None
self._http_client_sync: "httpx.Client | None" = None
self._vite_url: "str | None" = None
self._manifest: "dict[str, Any]" = {}
@property
def is_initialized(self) -> bool:
"""Whether the handler has been initialized.
Returns:
True when initialized, otherwise False.
"""
return self._initialized
[docs]
async def initialize_async(self, vite_url: "str | None" = None) -> None:
"""Initialize the handler asynchronously.
Args:
vite_url: Optional Vite server URL to use for proxying.
"""
if self._initialized:
return
if self._config.is_dev_mode and self._config.hot_reload:
self._init_http_clients(vite_url)
else:
await self._load_production_assets_async()
self._initialized = True
[docs]
def initialize_sync(self, vite_url: "str | None" = None) -> None:
"""Initialize the handler synchronously.
Args:
vite_url: Optional Vite server URL to use for proxying.
"""
if self._initialized:
return
if self._config.is_dev_mode and self._config.hot_reload:
self._init_http_clients(vite_url)
else:
self._load_production_assets_sync()
self._initialized = True
def _init_http_clients(self, vite_url: "str | None" = None) -> None:
"""Initialize HTTP clients for dev mode proxying."""
self._vite_url = vite_url or self._resolve_vite_url()
http2_enabled = self._config.http2
if http2_enabled:
try:
import h2 # noqa: F401 # pyright: ignore[reportMissingImports,reportUnusedImport]
except ImportError:
http2_enabled = False
self._http_client = httpx.AsyncClient(timeout=httpx.Timeout(5.0), http2=http2_enabled)
self._http_client_sync = httpx.Client(timeout=httpx.Timeout(5.0))
[docs]
async def shutdown_async(self) -> None:
"""Shutdown the handler asynchronously."""
if self._http_client is not None:
with suppress(RuntimeError):
await self._http_client.aclose()
self._http_client = None
if self._http_client_sync is not None:
with suppress(RuntimeError):
self._http_client_sync.close()
self._http_client_sync = None
def _load_production_assets_sync(self) -> None:
"""Load manifest and index.html synchronously in production modes."""
if self._config.mode != "external":
self._load_manifest_sync()
self._load_index_html_sync()
async def _load_production_assets_async(self) -> None:
"""Load manifest and index.html asynchronously in production modes."""
if self._config.mode != "external":
await self._load_manifest_async()
await self._load_index_html_async()
def _transform_html(
self, html: str, page_data: "dict[str, Any] | None" = None, csrf_token: "str | None" = None
) -> str:
"""Transform HTML by injecting CSRF token and/or page data.
Returns:
The transformed HTML.
"""
if self._spa_config is None:
if page_data is not None:
json_data = encode_json(page_data).decode("utf-8")
html = set_data_attribute(html, "#app", "data-page", json_data)
return html
if self._spa_config.inject_csrf and csrf_token:
script = f'window.{self._spa_config.csrf_var_name} = "{csrf_token}";'
html = inject_head_script(html, script, escape=False, nonce=self._config.csp_nonce)
if page_data is not None:
json_data = encode_json(page_data).decode("utf-8")
html = set_data_attribute(html, self._spa_config.app_selector, "data-page", json_data)
return html
async def _load_index_html_async(self) -> None:
"""Load and cache index.html asynchronously.
Raises:
ImproperlyConfiguredException: If index.html cannot be located.
"""
resolved_path: Path | None = None
for candidate in self._config.candidate_index_html_paths():
candidate_path = Path(candidate)
if candidate_path.exists():
resolved_path = candidate_path
break
if resolved_path is None:
self._raise_index_not_found()
raw_bytes = await anyio.Path(resolved_path).read_bytes()
html = raw_bytes.decode("utf-8")
html = self._transform_asset_urls_in_html(html)
self._cached_html = html
self._cached_bytes = html.encode("utf-8")
def _load_index_html_sync(self) -> None:
"""Load and cache index.html synchronously.
Raises:
ImproperlyConfiguredException: If index.html cannot be located.
"""
resolved_path: Path | None = None
for candidate in self._config.candidate_index_html_paths():
candidate_path = Path(candidate)
if candidate_path.exists():
resolved_path = candidate_path
break
if resolved_path is None:
self._raise_index_not_found()
raw_bytes = resolved_path.read_bytes()
html = raw_bytes.decode("utf-8")
html = self._transform_asset_urls_in_html(html)
self._cached_html = html
self._cached_bytes = html.encode("utf-8")
def _raise_index_not_found(self) -> NoReturn:
"""Raise an exception when index.html is not found.
Raises:
ImproperlyConfiguredException: Always raised.
"""
joined_paths = ", ".join(str(path) for path in self._config.candidate_index_html_paths())
msg = (
"index.html not found. "
f"Checked: {joined_paths}. "
"SPA mode requires index.html in one of the expected locations. "
"Did you forget to build your assets?"
)
raise ImproperlyConfiguredException(msg)
def _get_manifest_path(self) -> Path:
"""Get the path to the Vite manifest file.
Returns:
Absolute path to the manifest file location.
"""
bundle_dir = self._config.bundle_dir
if not bundle_dir.is_absolute():
bundle_dir = self._config.root_dir / bundle_dir
return bundle_dir / self._config.manifest_name
async def _load_manifest_async(self) -> None:
"""Asynchronously load the Vite manifest for asset URL transformation."""
manifest_path = anyio.Path(self._get_manifest_path())
try:
if await manifest_path.exists():
content = await manifest_path.read_bytes()
self._manifest = decode_json(content)
else:
logger.warning(
"Vite manifest not found at %s. "
"Asset URLs in index.html will not be transformed. "
"Run 'litestar assets build' to generate the manifest.",
manifest_path,
)
except OSError as exc:
logger.warning("Failed to read Vite manifest file: %s", exc)
except SerializationException as exc:
logger.warning("Failed to parse Vite manifest JSON: %s", exc)
def _load_manifest_sync(self) -> None:
"""Synchronously load the Vite manifest for asset URL transformation."""
manifest_path = self._get_manifest_path()
try:
if manifest_path.exists():
content = manifest_path.read_bytes()
self._manifest = decode_json(content)
else:
logger.warning(
"Vite manifest not found at %s. "
"Asset URLs in index.html will not be transformed. "
"Run 'litestar assets build' to generate the manifest.",
manifest_path,
)
except OSError as exc:
logger.warning("Failed to read Vite manifest file: %s", exc)
except SerializationException as exc:
logger.warning("Failed to parse Vite manifest JSON: %s", exc)
def _transform_asset_urls_in_html(self, html: str) -> str:
"""Transform source asset URLs to production hashed URLs using manifest.
Args:
html: The HTML to transform.
Returns:
The transformed HTML (or original HTML when no manifest is loaded).
"""
if not self._manifest:
return html
return transform_asset_urls(html, self._manifest, asset_url=self._config.asset_url, base_url=None)
def _get_csrf_token(self, request: "Request[Any, Any, Any]") -> "str | None":
"""Extract CSRF token from the request scope.
Args:
request: Incoming request.
Returns:
The CSRF token, or None if not present.
"""
from litestar.utils.empty import value_or_default
from litestar.utils.scope.state import ScopeState
return value_or_default(ScopeState.from_scope(request.scope).csrf_token, None)
[docs]
async def get_html(self, request: "Request[Any, Any, Any]", *, page_data: "dict[str, Any] | None" = None) -> str:
"""Get the HTML for the SPA with optional transformations.
Args:
request: Incoming request.
page_data: Optional page data to inject (e.g., Inertia page props).
Returns:
The rendered HTML.
Raises:
ImproperlyConfiguredException: If the handler is not initialized.
"""
if not self._initialized:
msg = "AppHandler not initialized. Call initialize() during app startup."
raise ImproperlyConfiguredException(msg)
needs_transform = self._spa_config is not None or page_data is not None
needs_csrf = self._spa_config is not None and self._spa_config.inject_csrf
csrf_token = self._get_csrf_token(request) if needs_csrf else None
if self._config.is_dev_mode and self._config.hot_reload:
html = await self._proxy_to_dev_server(request)
if needs_transform:
html = self._transform_html(html, page_data, csrf_token)
return html
if self._cached_html is None:
await self._load_index_html_async()
base_html = self._cached_html or ""
if not needs_transform:
return base_html
if page_data is not None or csrf_token is not None:
return self._transform_html(base_html, page_data, csrf_token)
if self._spa_config is not None and self._spa_config.cache_transformed_html:
if self._cached_transformed_html is not None:
return self._cached_transformed_html
self._cached_transformed_html = self._transform_html(base_html, None, None)
return self._cached_transformed_html
return self._transform_html(base_html, None, None)
[docs]
def get_html_sync(self, *, page_data: "dict[str, Any] | None" = None, csrf_token: "str | None" = None) -> str:
"""Get the HTML for the SPA synchronously.
Args:
page_data: Optional page data to inject (e.g., Inertia page props).
csrf_token: Optional CSRF token to inject.
Returns:
The rendered HTML.
Raises:
ImproperlyConfiguredException: If dev-mode HTTP clients are not initialized or Vite URL is unresolved.
"""
if not self._initialized:
logger.warning(
"AppHandler lazy init triggered - lifespan may not have run. "
"Consider calling initialize_sync() explicitly during app startup."
)
self.initialize_sync()
needs_transform = self._spa_config is not None or page_data is not None
if not needs_transform:
if self._config.is_dev_mode and self._config.hot_reload:
return self._proxy_to_dev_server_sync()
return self._cached_html or ""
if self._config.is_dev_mode and self._config.hot_reload:
html = self._proxy_to_dev_server_sync()
return self._transform_html(html, page_data, csrf_token)
base_html = self._cached_html or ""
return self._transform_html(base_html, page_data, csrf_token)
[docs]
async def get_bytes(self) -> bytes:
"""Get cached index.html bytes (production).
Returns:
Cached HTML bytes.
Raises:
ImproperlyConfiguredException: If index.html cannot be located.
"""
if not self._initialized:
logger.warning(
"AppHandler lazy init triggered - lifespan may not have run. "
"Consider calling initialize_sync() explicitly during app startup."
)
await self.initialize_async()
if self._cached_bytes is None:
await self._load_index_html_async()
return self._cached_bytes or b""
async def _proxy_to_dev_server(self, request: "Request[Any, Any, Any]") -> str:
"""Proxy request to Vite dev server and return HTML.
Args:
request: Incoming request.
Returns:
HTML from the Vite dev server.
Raises:
ImproperlyConfiguredException: If the dev server cannot be reached or proxying is not configured.
"""
if self._http_client is None:
msg = "HTTP client not initialized. Ensure initialize_async() was called for dev mode."
raise ImproperlyConfiguredException(msg)
if self._vite_url is None:
msg = "Vite URL not resolved. Ensure initialize_sync() or initialize_async() was called."
raise ImproperlyConfiguredException(msg)
target_url = f"{self._vite_url}/"
try:
response = await self._http_client.get(target_url, follow_redirects=True)
response.raise_for_status()
except httpx.HTTPError as e:
msg = f"Failed to proxy request to Vite server at {target_url}. Is the dev server running? Error: {e!s}"
raise ImproperlyConfiguredException(msg) from e
else:
return response.text
def _proxy_to_dev_server_sync(self) -> str:
"""Proxy request to Vite dev server synchronously.
Returns:
HTML from the Vite dev server.
Raises:
ImproperlyConfiguredException: If the dev server cannot be reached or proxying is not configured.
"""
if self._http_client_sync is None:
msg = "HTTP client not initialized. Ensure initialize_sync() was called for dev mode."
raise ImproperlyConfiguredException(msg)
if self._vite_url is None:
msg = "Vite URL not resolved. Ensure initialize_sync() or initialize_async() was called."
raise ImproperlyConfiguredException(msg)
target_url = f"{self._vite_url}/"
try:
response = self._http_client_sync.get(target_url, follow_redirects=True)
response.raise_for_status()
except httpx.HTTPError as e:
msg = f"Failed to proxy request to Vite server at {target_url}. Is the dev server running? Error: {e!s}"
raise ImproperlyConfiguredException(msg) from e
else:
return response.text
def _resolve_vite_url(self) -> str:
"""Resolve the Vite server URL from hotfile or config.
Returns:
The base Vite URL without a trailing slash.
"""
hotfile = self._config.bundle_dir / self._config.hot_file
if not hotfile.is_absolute():
hotfile = self._config.root_dir / hotfile
if hotfile.exists():
try:
url = hotfile.read_text().strip()
if url:
return url.rstrip("/")
except OSError:
pass
return f"{self._config.protocol}://{self._config.host}:{self._config.port}"
[docs]
def create_route_handler(self) -> Any:
"""Create a Litestar route handler for the SPA.
Returns:
A Litestar route handler suitable for registering on an application.
"""
is_dev = self._config.is_dev_mode and self._config.hot_reload
opt: dict[str, Any] = {}
if self._config.exclude_static_from_auth:
opt["exclude_from_auth"] = True
opt["_vite_spa_handler"] = self
guards: "list[Guard] | None" = list(self._config.guards) if self._config.guards else None # pyright: ignore[reportUnknownVariableType,reportUnknownMemberType,reportUnknownArgumentType]
asset_url = self._config.asset_url
spa_path = self._config.spa_path
effective_spa_path = spa_path if spa_path is not None else "/"
include_root = self._config.include_root_spa_paths
if effective_spa_path and effective_spa_path != "/":
base = effective_spa_path.rstrip("/")
paths: list[str] = [f"{base}/", f"{base}/{{path:path}}"]
if include_root:
paths.extend(["/", "/{path:path}"])
else:
paths = ["/", "/{path:path}"]
needs_exclusion = asset_url and asset_url != "/" and (effective_spa_path == "/" or include_root)
asset_prefix = asset_url.rstrip("/") if needs_exclusion else None
if asset_prefix:
opt["_vite_asset_prefix"] = asset_prefix
if is_dev:
return get(path=paths, name="vite_spa", opt=opt, include_in_schema=False, guards=guards)(spa_handler_dev)
return get(path=paths, name="vite_spa", opt=opt, include_in_schema=False, cache=3600, guards=guards)(
spa_handler_prod
)
litestar-vite