"""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)
- asyncapi.json (AsyncAPI schema when AsyncAPIPlugin is registered)
Both CLI and Plugin should call this function to guarantee byte-identical output.
"""
import warnings
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 schema dict (for downstream use)."""
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,
))
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]
def _invalidate_missing_asyncapi_outputs(app: "Litestar", types_config: "TypeGenConfig") -> None:
"""Warn about missing schema providers and remove obsolete generated contracts."""
stale_paths = [
types_config.asyncapi_path or types_config.output / "asyncapi.json",
types_config.channels_ts_path or types_config.output / "channels.ts",
]
stale = any(path.exists() for path in stale_paths)
from litestar.channels import ChannelsPlugin
from litestar.response import ServerSentEvent
from litestar.routes import HTTPRoute, WebSocketRoute
has_realtime = any(isinstance(plugin, ChannelsPlugin) for plugin in app.plugins)
for route in app.routes:
if isinstance(route, WebSocketRoute):
has_realtime = True
elif isinstance(route, HTTPRoute):
for handler in route.route_handlers:
annotation = handler.parsed_return_field.annotation
if isinstance(annotation, type) and issubclass(annotation, ServerSentEvent):
has_realtime = True
if stale or has_realtime:
warnings.warn(
"Channel type generation requires litestar-asyncapi and a registered AsyncAPIPlugin. "
"Stale asyncapi.json and channels.ts outputs are removed when the plugin is absent.",
UserWarning,
stacklevel=2,
)
for path in stale_paths:
path.unlink(missing_ok=True)
[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.
OpenAPI schema, route metadata, route definitions, and Inertia page prop
artifacts require an active OpenAPIPlugin configuration. AsyncAPI export
requires a registered AsyncAPIPlugin (litestar-asyncapi) on the application
and generate_channels=True.
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 and AsyncAPIPlugin is registered)
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._asyncapi import find_asyncapi_plugin
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
has_asyncapi = find_asyncapi_plugin(app) is not None
if types_config.generate_channels and not has_asyncapi:
_invalidate_missing_asyncapi_outputs(app, types_config)
if not has_openapi:
if types_config.generate_channels and has_asyncapi:
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 has_asyncapi:
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 schema to file when AsyncAPIPlugin is registered.
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
schema_dict = resolve_asyncapi_document(app)
if schema_dict is None:
return
asyncapi_path = types_config.asyncapi_path
if asyncapi_path is None:
asyncapi_path = types_config.output / "asyncapi.json"
result.asyncapi_schema = schema_dict
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")
