"""Unified asset export pipeline for deterministic code generation.
This module provides a single entry point for exporting all integration artifacts:
- openapi.json (OpenAPI schema with Inertia types registered)
- routes.json (route metadata)
- routes.ts (Ziggy-style typed routes)
- inertia-pages.json (Inertia page props metadata)
Both CLI and Plugin should call this function to guarantee byte-identical output.
"""
import contextlib
from dataclasses import dataclass, field
from functools import partial
from importlib.metadata import PackageNotFoundError, version
from pathlib import Path
from typing import TYPE_CHECKING, Any
if TYPE_CHECKING:
from collections.abc import Callable
from litestar import Litestar
from litestar_vite.config import TypeGenConfig, ViteConfig
[docs]
@dataclass
class ExportResult:
"""Result of the export operation."""
exported_files: list[str] = field(default_factory=list) # pyright: ignore[reportUnknownVariableType]
"""Files that were written (content changed)."""
unchanged_files: list[str] = field(default_factory=list) # pyright: ignore[reportUnknownVariableType]
"""Files that were skipped (content unchanged)."""
openapi_schema: "dict[str, Any] | None" = None
"""The OpenAPI schema dict (for downstream use)."""
asyncapi_schema: "dict[str, Any] | None" = None
"""The AsyncAPI 3.0 schema dict (for downstream use)."""
asyncapi_source: "str | None" = None
"""The authoritative source for the AsyncAPI schema ('litestar-asyncapi' or 'builtin')."""
def fmt_path(path: Path) -> str:
"""Format path for display, using relative path when possible.
Returns:
The result.
"""
try:
return str(path.relative_to(Path.cwd()))
except ValueError:
return str(path)
[docs]
def typegen_outputs_requested(types_config: "TypeGenConfig") -> bool:
"""Return whether the typegen config asks for any generated artifact."""
return any((
types_config.generate_sdk,
types_config.generate_zod,
types_config.generate_schemas,
types_config.generate_routes,
types_config.generate_page_props,
types_config.generate_channels,
))
[docs]
def app_has_realtime_surface(app: "Litestar") -> bool:
"""Return whether the Litestar application defines any realtime surface.
A realtime surface includes:
- Any plugin in ``app.plugins`` that provides AsyncAPI schemas.
- Any route in ``app.routes`` that is a ``litestar.routes.WebSocketRoute``.
- Any plugin in ``app.plugins`` that is a ``litestar.channels.ChannelsPlugin``.
- Any HTTP route handler whose return annotation represents a Server-Sent Event (SSE).
Args:
app: The Litestar application instance.
Returns:
True if any realtime route, plugin, or SSE handler is detected, otherwise False.
"""
from litestar_vite.codegen._asyncapi import find_asyncapi_plugin
if find_asyncapi_plugin(app) is not None:
return True
from litestar.channels import ChannelsPlugin
plugins = getattr(app, "plugins", None)
if plugins is not None and hasattr(plugins, "get"):
with contextlib.suppress(KeyError, AttributeError):
if plugins.get(ChannelsPlugin) is not None:
return True
with contextlib.suppress(KeyError, AttributeError):
if plugins.get("ChannelsPlugin") is not None:
return True
from litestar.routes import HTTPRoute, WebSocketRoute
for route in app.routes:
if isinstance(route, WebSocketRoute):
return True
from litestar_vite.codegen._asyncapi import _is_sse_type # pyright: ignore[reportPrivateUsage]
for route in app.routes:
if not isinstance(route, HTTPRoute):
continue
for handler in route.route_handlers:
return_field = getattr(handler, "parsed_return_field", None)
annotation = getattr(return_field, "annotation", None)
if annotation is None:
annotation = getattr(handler, "return_type", None)
if _is_sse_type(annotation):
return True
return False
def _resolve_serializer(
app: "Litestar", serializer: "Callable[[Any], bytes] | None" = None
) -> "Callable[[Any], bytes]":
"""Resolve a JSON serializer using application type encoders when none is provided.
Args:
app: The Litestar application instance.
serializer: An optional custom serializer function.
Returns:
A callable serializer for encoding JSON.
"""
if serializer is not None:
return serializer
encoders: Any
try:
encoders = app.type_encoders # pyright: ignore[reportUnknownMemberType]
except AttributeError:
encoders = None
from litestar.serialization import encode_json, get_serializer
return partial(encode_json, serializer=get_serializer(encoders if isinstance(encoders, dict) else None)) # pyright: ignore[reportUnknownArgumentType]
[docs]
def export_integration_assets(
app: "Litestar", config: "ViteConfig", *, serializer: "Callable[[Any], bytes] | None" = None
) -> ExportResult:
"""Export all integration artifacts with deterministic output.
This is the single source of truth for code generation. Both CLI commands
and Plugin startup should call this function to ensure byte-identical output.
AsyncAPI export is independent of OpenAPI availability when channels are
enabled and a realtime surface is present on the application (WebSocket routes,
ChannelsPlugin, or SSE handlers). The configuration flag generate_channels=True
is an opt-in ceiling rather than a mandate, so a REST-only application produces
no AsyncAPI artifacts even when generate_channels is true. OpenAPI schema, route
metadata, route definitions, and Inertia page prop artifacts require an active
OpenAPI configuration.
The export order is critical:
1. Register Inertia page prop types in OpenAPI schema (mutates schema_dict)
2. Export openapi.json (now includes session prop types)
3. Export routes.json (uses schema for component refs)
4. Export routes.ts (if enabled)
5. Export inertia-pages.json (if enabled)
6. Export asyncapi.json (if enabled)
Args:
app: The Litestar application instance.
config: The ViteConfig instance.
serializer: Optional custom serializer for OpenAPI schema encoding.
Returns:
ExportResult with lists of exported and unchanged files.
"""
from litestar._openapi.plugin import OpenAPIPlugin
from litestar_vite.codegen._inertia import generate_inertia_pages_json
from litestar_vite.codegen._routes import extract_route_metadata
from litestar_vite.config import InertiaConfig, InertiaTypeGenConfig, TypeGenConfig
result = ExportResult()
if not isinstance(config.types, TypeGenConfig):
return result
types_config = config.types
if not typegen_outputs_requested(types_config):
return result
from litestar.exceptions import ImproperlyConfiguredException
plugins = getattr(app, "plugins", ())
openapi_plugin = next((p for p in plugins if isinstance(p, OpenAPIPlugin)), None)
has_openapi = False
if openapi_plugin is not None:
try:
_ = openapi_plugin.openapi_config
has_openapi = True
except ImproperlyConfiguredException:
has_openapi = False
if not has_openapi:
if types_config.generate_channels and app_has_realtime_surface(app):
export_asyncapi(
app=app, types_config=types_config, serializer=_resolve_serializer(app, serializer), result=result
)
return result
serializer = _resolve_serializer(app, serializer)
schema_dict = app.openapi_schema.to_schema()
inertia_pages_data: dict[str, Any] | None = None
if isinstance(config.inertia, InertiaConfig) and types_config.generate_page_props:
inertia_type_gen = config.inertia.type_gen or InertiaTypeGenConfig()
inertia_pages_data = generate_inertia_pages_json(
app,
openapi_schema=schema_dict,
include_default_auth=inertia_type_gen.include_default_auth,
include_default_flash=inertia_type_gen.include_default_flash,
inertia_config=config.inertia,
types_config=types_config,
)
result.openapi_schema = schema_dict
export_openapi(schema_dict=schema_dict, types_config=types_config, serializer=serializer, result=result)
routes_metadata = extract_route_metadata(app, openapi_schema=schema_dict)
export_routes_json(
app=app, types_config=types_config, openapi_schema=schema_dict, routes_metadata=routes_metadata, result=result
)
if types_config.generate_routes:
export_routes_ts(
app=app,
types_config=types_config,
openapi_schema=schema_dict,
routes_metadata=routes_metadata,
result=result,
)
if (
isinstance(config.inertia, InertiaConfig)
and types_config.generate_page_props
and types_config.page_props_path
and inertia_pages_data is not None
):
export_inertia_pages(pages_data=inertia_pages_data, types_config=types_config, result=result)
if types_config.generate_channels and app_has_realtime_surface(app):
export_asyncapi(app=app, types_config=types_config, serializer=serializer, result=result)
return result
def export_openapi(
*,
schema_dict: "dict[str, Any]",
types_config: "TypeGenConfig",
serializer: "Callable[[Any], bytes]",
result: ExportResult,
) -> None:
"""Export OpenAPI schema to file."""
from litestar_vite.codegen._utils import encode_deterministic_json, write_if_changed
openapi_path = types_config.openapi_path
if openapi_path is None:
openapi_path = types_config.output / "openapi.json"
schema_content = encode_deterministic_json(schema_dict, serializer=serializer)
if write_if_changed(openapi_path, schema_content):
result.exported_files.append(f"openapi: {fmt_path(openapi_path)}")
else:
result.unchanged_files.append("openapi.json")
def export_routes_json(
*,
app: "Litestar",
types_config: "TypeGenConfig",
openapi_schema: "dict[str, Any]",
routes_metadata: "list[Any]",
result: ExportResult,
) -> None:
"""Export routes metadata to JSON file."""
from litestar_vite.codegen._routes import generate_routes_json
from litestar_vite.codegen._utils import encode_deterministic_json, write_if_changed
try:
litestar_version = version("litestar")
except PackageNotFoundError:
litestar_version = "unknown"
routes_path = types_config.routes_path
if routes_path is None:
routes_path = types_config.output / "routes.json"
routes_data = generate_routes_json(
app, include_components=True, openapi_schema=openapi_schema, routes_metadata=routes_metadata
)
routes_data["litestar_version"] = litestar_version
routes_content = encode_deterministic_json(routes_data)
if write_if_changed(routes_path, routes_content):
result.exported_files.append(fmt_path(routes_path))
else:
result.unchanged_files.append("routes.json")
def export_routes_ts(
*,
app: "Litestar",
types_config: "TypeGenConfig",
openapi_schema: "dict[str, Any]",
routes_metadata: "list[Any]",
result: ExportResult,
) -> None:
"""Export typed routes TypeScript file."""
from litestar_vite.codegen._routes import generate_routes_ts
from litestar_vite.codegen._utils import write_if_changed
routes_ts_path = types_config.routes_ts_path
if routes_ts_path is None:
routes_ts_path = types_config.output / "routes.ts"
routes_ts_content = generate_routes_ts(
app, openapi_schema=openapi_schema, global_route=types_config.global_route, routes_metadata=routes_metadata
)
if write_if_changed(routes_ts_path, routes_ts_content):
result.exported_files.append(fmt_path(routes_ts_path))
else:
result.unchanged_files.append("routes.ts")
def export_inertia_pages(*, pages_data: "dict[str, Any]", types_config: "TypeGenConfig", result: ExportResult) -> None:
"""Export Inertia pages metadata to JSON file."""
from litestar_vite.codegen._utils import encode_deterministic_json, write_if_changed
page_props_path = types_config.page_props_path
if page_props_path is None:
return
pages_content = encode_deterministic_json(pages_data)
if write_if_changed(page_props_path, pages_content):
result.exported_files.append(fmt_path(page_props_path))
else:
result.unchanged_files.append("inertia-pages.json")
[docs]
def export_asyncapi(
*,
app: "Litestar",
types_config: "TypeGenConfig",
serializer: "Callable[[Any], bytes] | None" = None,
result: ExportResult,
) -> None:
"""Export AsyncAPI 3.0 schema to file.
Args:
app: The Litestar application instance.
types_config: The type generation configuration.
serializer: Optional custom serializer for JSON encoding.
result: ExportResult accumulator for exported or unchanged files.
"""
from litestar_vite.codegen._asyncapi import resolve_asyncapi_document
from litestar_vite.codegen._utils import encode_deterministic_json, write_if_changed
asyncapi_path = types_config.asyncapi_path
if asyncapi_path is None:
asyncapi_path = types_config.output / "asyncapi.json"
schema_dict, source = resolve_asyncapi_document(app)
result.asyncapi_schema = schema_dict
result.asyncapi_source = source
schema_content = encode_deterministic_json(schema_dict, serializer=serializer)
if write_if_changed(asyncapi_path, schema_content):
result.exported_files.append(f"asyncapi: {fmt_path(asyncapi_path)}")
else:
result.unchanged_files.append("asyncapi.json")
