Source code for litestar_vite.codegen._inertia

"""Inertia page-props metadata extraction and export."""

import re
from collections.abc import Mapping
from contextlib import suppress
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING, Any, cast

from litestar._openapi.datastructures import _get_normalized_schema_key  # pyright: ignore[reportPrivateUsage]
from litestar.handlers import HTTPRouteHandler
from litestar.openapi.spec import Reference, Schema
from litestar.response.base import ASGIResponse
from litestar.routes import HTTPRoute
from litestar.types.builtin_types import NoneType
from litestar.typing import FieldDefinition

from litestar_vite.codegen._openapi import (
    OpenAPISupport,
    build_schema_name_map,
    merge_generated_components_into_openapi,
    openapi_components_schemas,
    resolve_page_props_field_definition,
    schema_name_from_ref,
)
from litestar_vite.codegen._ts import collect_ref_names, normalize_path, python_type_to_typescript, ts_type_from_openapi

if TYPE_CHECKING:
    from litestar import Litestar

    from litestar_vite.config import InertiaConfig, TypeGenConfig

# Compiled regex for splitting TypeScript type strings on union/intersection operators
_TYPE_OPERATOR_RE = re.compile(r"(\s*[|&]\s*)")


def str_list_factory() -> list[str]:
    """Return an empty ``list[str]`` (typed for pyright).

    Returns:
        An empty list.
    """
    return []


def _pick_inertia_method(http_methods: "set[Any] | frozenset[Any] | None") -> str:
    """Pick a deterministic HTTP method for Inertia page props inference.

    Inertia pages are typically loaded via GET requests, so we prefer GET.
    For determinism, we sort remaining methods alphabetically.

    Args:
        http_methods: Set of HTTP methods from the route handler.

    Returns:
        The selected HTTP method string.
    """
    if not http_methods:
        return "GET"
    # Prefer GET for Inertia page loads
    if "GET" in http_methods:
        return "GET"
    # Fallback to alphabetically first method for determinism
    return min(http_methods)


def normalize_type_name(type_name: str, openapi_schemas: set[str]) -> str:
    """Strip module prefix from mangled type names.

    Always converts 'app_lib_schema_NoProps' -> 'NoProps' because:
    1. If 'NoProps' exists in OpenAPI, it will be imported correctly
    2. If 'NoProps' doesn't exist, the error message is clearer for users
       (they can add it to OpenAPI or configure type_import_paths)

    The mangled name 'app_lib_schema_NoProps' will NEVER work - it doesn't
    exist anywhere. The short name is always preferable.

    Args:
        type_name: The potentially mangled type name.
        openapi_schemas: Set of available OpenAPI schema names.

    Returns:
        The normalized (unmangled) type name.
    """
    if type_name in openapi_schemas:
        return type_name

    # Check if this looks like a mangled module path (contains underscores)
    if "_" not in type_name:
        return type_name

    # Try progressively shorter suffixes to find the class name
    parts = type_name.split("_")
    for i in range(len(parts)):
        short_name = "_".join(parts[i:])
        # Prefer OpenAPI match, but if we get to the last part, use it anyway
        if short_name in openapi_schemas:
            return short_name

    # Use the last part as the class name (e.g., 'NoProps' from 'app_lib_schema_NoProps')
    # This is always better than the mangled name for error messages
    return parts[-1] if parts else type_name


def normalize_type_string(type_string: str, openapi_schemas: set[str]) -> str:
    """Normalize all type names within a TypeScript type string.

    Handles union types like 'any | app_lib_schema_NoProps' by parsing the
    string and normalizing each type name individually.

    Args:
        type_string: A TypeScript type string (may contain unions, intersections).
        openapi_schemas: Set of available OpenAPI schema names.

    Returns:
        The type string with all type names normalized.
    """
    # Primitives and special types that should not be normalized
    skip_types = {"any", "unknown", "null", "undefined", "void", "never", "string", "number", "boolean", "object"}

    # Split on | and & while preserving whitespace
    tokens = _TYPE_OPERATOR_RE.split(type_string)
    result_parts: list[str] = []

    for token in tokens:
        stripped = token.strip()
        # Keep operators and whitespace as-is
        if stripped in {"|", "&", ""} or stripped in skip_types or stripped == "{}":
            result_parts.append(token)
        # Normalize type names
        else:
            normalized = normalize_type_name(stripped, openapi_schemas)
            # Preserve original whitespace around the type
            prefix = token[: len(token) - len(token.lstrip())]
            suffix = token[len(token.rstrip()) :]
            result_parts.append(prefix + normalized + suffix)

    return "".join(result_parts)


[docs] @dataclass class InertiaPageMetadata: """Metadata for a single Inertia page component.""" component: str route_path: str props_type: str | None = None schema_ref: str | None = None handler_name: str | None = None ts_type: str | None = None custom_types: list[str] = field(default_factory=str_list_factory) wrap_with_content: bool = False
def get_return_type_name(handler: HTTPRouteHandler) -> "str | None": field_definition = handler.parsed_fn_signature.return_type excluded_types: tuple[type[Any], ...] = (NoneType, ASGIResponse) if field_definition.is_subclass_of(excluded_types): return None fn = handler.fn with suppress(AttributeError): return_annotation = fn.__annotations__.get("return") if isinstance(return_annotation, str) and return_annotation: return return_annotation raw = field_definition.raw if isinstance(raw, str): return raw if isinstance(raw, type): return raw.__name__ origin: Any = None with suppress(AttributeError): origin = field_definition.origin if isinstance(origin, type): return origin.__name__ return str(raw) def _filter_page_props_response_types(field_definition: FieldDefinition) -> FieldDefinition | None: """Filter response wrapper types out of a handler return annotation.""" from typing import Union if not field_definition.is_union: return field_definition props_types: list[type[Any]] = [] for inner in field_definition.inner_types: if inner.is_subclass_of(NoneType): continue if inner.is_subclass_of(ASGIResponse): continue props_types.append(cast("type[Any]", inner.annotation)) if not props_types: return None if len(props_types) == 1: return FieldDefinition.from_annotation(props_types[0]) props_types.sort(key=lambda t: getattr(t, "__qualname__", str(t))) union_type = Union[tuple(props_types)] # type: ignore[valid-type] # noqa: UP007 return FieldDefinition.from_annotation(union_type) def should_wrap_page_props_with_content(handler: HTTPRouteHandler) -> bool: """Return whether generated props should stay nested under ``content``. Runtime Inertia responses flatten mapping payloads into top-level props, but keep non-mapping objects nested under ``content``. The generated metadata has to preserve that distinction so ``page-props.ts`` matches actual responses. """ field_definition = _filter_page_props_response_types(handler.parsed_fn_signature.return_type) if field_definition is None: return False if field_definition.is_subclass_of((NoneType, ASGIResponse)): return False return not field_definition.is_subclass_of(Mapping) def get_openapi_schema_ref( handler: HTTPRouteHandler, openapi_schema: dict[str, Any] | None, route_path: str, method: str = "GET" ) -> "str | None": if not openapi_schema: return None paths = openapi_schema.get("paths", {}) path_item = paths.get(route_path, {}) operation = path_item.get(method.lower(), {}) responses = operation.get("responses", {}) success_response = responses.get("200", responses.get("2XX", {})) content = success_response.get("content", {}) json_content = content.get("application/json", {}) schema = json_content.get("schema", {}) ref = schema.get("$ref") return cast("str | None", ref) if ref else None def extract_inertia_component(handler: HTTPRouteHandler) -> str | None: opt = handler.opt or {} component = opt.get("component") or opt.get("page") return component if isinstance(component, str) and component else None def infer_inertia_props_type( component: str, handler: HTTPRouteHandler, schema_creator: Any, page_schema_keys: dict[str, tuple[str, ...]], page_schema_dicts: dict[str, dict[str, Any]], *, fallback_type: str, ) -> str | None: if schema_creator is not None: field_def, schema_result = resolve_page_props_field_definition(handler, schema_creator) if field_def is not None and isinstance(schema_result, Reference): page_schema_keys[component] = _get_normalized_schema_key(field_def) return None if isinstance(schema_result, Schema): schema_dict = schema_result.to_schema() page_schema_dicts[component] = schema_dict return ts_type_from_openapi(schema_dict) return None raw_type = get_return_type_name(handler) if not raw_type: return None props_type, _ = python_type_to_typescript(raw_type, fallback=fallback_type) return props_type def finalize_inertia_pages( pages: list[InertiaPageMetadata], *, openapi_support: OpenAPISupport, page_schema_keys: dict[str, tuple[str, ...]], page_schema_dicts: dict[str, dict[str, Any]], ) -> None: context = openapi_support.context if context is None: return generated_components = context.schema_registry.generate_components_schemas() name_map = build_schema_name_map(context.schema_registry) openapi_components = openapi_components_schemas(openapi_support.openapi_schema) # Build set of available OpenAPI schema names for type normalization openapi_schema_names: set[str] = set(openapi_components.keys()) openapi_schema_names.update(generated_components.keys()) if openapi_support.openapi_schema is not None: merge_generated_components_into_openapi(openapi_support.openapi_schema, generated_components) for page in pages: schema_key = page_schema_keys.get(page.component) schema_name: str | None = None if page.schema_ref: schema_name = schema_name_from_ref(page.schema_ref) elif schema_key: schema_name = name_map.get(schema_key) if schema_name: # Normalize mangled type names (e.g., 'app_lib_schema_NoProps' -> 'NoProps') normalized_name = normalize_type_name(schema_name, openapi_schema_names) page.ts_type = normalized_name page.props_type = normalized_name elif page.props_type: # Normalize type names in union/intersection type strings # (e.g., 'any | app_lib_schema_NoProps' -> 'any | NoProps') page.props_type = normalize_type_string(page.props_type, openapi_schema_names) custom_types: set[str] = set() if page.ts_type: custom_types.add(page.ts_type) if page.schema_ref: openapi_schema_dict = openapi_components.get(page.ts_type or "") if isinstance(openapi_schema_dict, dict): custom_types.update(collect_ref_names(openapi_schema_dict)) else: page_schema_dict = page_schema_dicts.get(page.component) if isinstance(page_schema_dict, dict): custom_types.update(collect_ref_names(page_schema_dict)) elif schema_key: registered = context.schema_registry._schema_key_map.get( # pyright: ignore[reportPrivateUsage] schema_key ) if registered: custom_types.update(collect_ref_names(registered.schema.to_schema())) # Normalize all custom type names page.custom_types = sorted(normalize_type_name(t, openapi_schema_names) for t in custom_types)
[docs] def extract_inertia_pages( app: "Litestar", *, openapi_schema: dict[str, Any] | None = None, fallback_type: "str" = "unknown", openapi_support: OpenAPISupport | None = None, ) -> list[InertiaPageMetadata]: """Extract Inertia page metadata from an application. When multiple handlers map to the same component, GET handlers are preferred since Inertia pages are typically loaded via GET requests. Args: app: Litestar application instance. openapi_schema: Optional OpenAPI schema dict. fallback_type: TypeScript fallback type for unknown types. openapi_support: Optional shared OpenAPISupport instance. If not provided, a new one will be created. Sharing improves determinism and performance. Returns: List of InertiaPageMetadata for each discovered page. """ # Track seen components: component -> (metadata, is_get_handler) # When multiple handlers map to the same component, prefer GET handlers seen_components: dict[str, tuple[InertiaPageMetadata, bool]] = {} if openapi_support is None: openapi_support = OpenAPISupport.from_app(app, openapi_schema) page_schema_keys: dict[str, tuple[str, ...]] = {} page_schema_dicts: dict[str, dict[str, Any]] = {} for http_route, route_handler in iter_route_handlers(app): component = extract_inertia_component(route_handler) if not component: continue normalized_path = normalize_path(str(http_route.path)) handler_name = route_handler.handler_name or route_handler.name is_get_handler = "GET" in (route_handler.http_methods or set()) props_type = infer_inertia_props_type( component, route_handler, openapi_support.schema_creator, page_schema_keys, page_schema_dicts, fallback_type=fallback_type, ) method = _pick_inertia_method(route_handler.http_methods) schema_ref = get_openapi_schema_ref(route_handler, openapi_schema, normalized_path, method=str(method)) page_metadata = InertiaPageMetadata( component=component, route_path=normalized_path, props_type=props_type, schema_ref=schema_ref, handler_name=handler_name, wrap_with_content=should_wrap_page_props_with_content(route_handler), ) # Prefer GET handlers when multiple handlers map to the same component existing = seen_components.get(component) if existing is None: seen_components[component] = (page_metadata, is_get_handler) elif is_get_handler and not existing[1]: # New handler is GET, existing is not - prefer the GET handler seen_components[component] = (page_metadata, is_get_handler) # Otherwise keep existing (it's either GET or we prefer first-seen for determinism) pages = [entry[0] for entry in seen_components.values()] if openapi_support.enabled: finalize_inertia_pages( pages, openapi_support=openapi_support, page_schema_keys=page_schema_keys, page_schema_dicts=page_schema_dicts, ) return pages
def iter_route_handlers(app: "Litestar") -> "list[tuple[HTTPRoute, HTTPRouteHandler]]": """Iterate over HTTP route handlers in an app. Returns a deterministically sorted list to ensure consistent output across multiple runs. Handlers are sorted by (route_path, handler_name). Returns: A list of (http_route, route_handler) tuples, sorted for determinism. """ handlers: list[tuple[HTTPRoute, HTTPRouteHandler]] = [] for route in app.routes: if isinstance(route, HTTPRoute): handlers.extend((route, route_handler) for route_handler in route.route_handlers) # Sort by route path, then handler name for deterministic ordering return sorted(handlers, key=lambda x: (str(x[0].path), x[1].handler_name or x[1].name or "")) def get_fallback_ts_type(types_config: "TypeGenConfig | None") -> str: fallback_type = types_config.fallback_type if types_config is not None else "unknown" return "any" if fallback_type == "any" else "unknown" def ts_type_from_value(value: Any, *, fallback_ts_type: str) -> str: ts_type = fallback_ts_type if value is None: ts_type = "null" elif isinstance(value, bool): ts_type = "boolean" elif isinstance(value, str): ts_type = "string" elif isinstance(value, (int, float)): ts_type = "number" elif isinstance(value, (bytes, bytearray, Path)): ts_type = "string" elif isinstance(value, (list, tuple, set, frozenset)): ts_type = f"{fallback_ts_type}[]" elif isinstance(value, dict): ts_type = f"Record<string, {fallback_ts_type}>" return ts_type def should_register_value_schema(value: Any) -> bool: if value is None: return False return not isinstance(value, (bool, str, int, float, bytes, bytearray, Path, list, tuple, set, frozenset, dict)) def process_session_props( session_props: "set[str] | dict[str, type]", shared_props: dict[str, dict[str, Any]], shared_schema_keys: dict[str, tuple[str, ...]], openapi_support: OpenAPISupport, fallback_ts_type: str, ) -> None: """Process session props and add them to shared_props. Handles both set[str] (legacy) and dict[str, type] (new typed) formats. """ if isinstance(session_props, dict): # New behavior: dict maps prop names to Python types for key, prop_type_class in session_props.items(): if not key: continue # Register the type with OpenAPI if possible if openapi_support.enabled and openapi_support.schema_creator: try: field_def = FieldDefinition.from_annotation(prop_type_class) schema_result = openapi_support.schema_creator.for_field_definition(field_def) if isinstance(schema_result, Reference): shared_schema_keys[key] = _get_normalized_schema_key(field_def) type_name = prop_type_class.__name__ if hasattr(prop_type_class, "__name__") else fallback_ts_type shared_props.setdefault(key, {"type": type_name, "optional": True}) except (AttributeError, TypeError, ValueError): # pragma: no cover - defensive shared_props.setdefault(key, {"type": fallback_ts_type, "optional": True}) else: type_name = prop_type_class.__name__ if hasattr(prop_type_class, "__name__") else fallback_ts_type shared_props.setdefault(key, {"type": type_name, "optional": True}) else: # Legacy behavior: set of prop names (types are unknown) for key in session_props: if not key: continue shared_props.setdefault(key, {"type": fallback_ts_type, "optional": True}) def process_shared_prop_types( shared_prop_types: "dict[str, Any]", shared_props: dict[str, dict[str, Any]], shared_schema_keys: dict[str, tuple[str, ...]], shared_schema_dicts: dict[str, dict[str, Any]], openapi_support: OpenAPISupport, fallback_ts_type: str, ) -> None: """Register declared ``share()`` prop annotations. Named models register as references and resolve to their generated name later. Containers and unions have no component of their own, so their inline schema is kept and rendered once nested references can be resolved. Unlike session props these overwrite any generated default, because an explicit Python annotation is more accurate than the synthesized convention it replaces. """ for key, annotation in shared_prop_types.items(): if not key: continue type_name = fallback_ts_type if openapi_support.enabled and openapi_support.schema_creator: try: field_def = FieldDefinition.from_annotation(annotation) schema_result = openapi_support.schema_creator.for_field_definition(field_def) if isinstance(schema_result, Reference): shared_schema_keys[key] = _get_normalized_schema_key(field_def) type_name = getattr(annotation, "__name__", fallback_ts_type) else: shared_schema_dicts[key] = schema_result.to_schema() except (AttributeError, TypeError, ValueError): # pragma: no cover - defensive type_name = fallback_ts_type else: type_name = getattr(annotation, "__name__", fallback_ts_type) shared_props[key] = {"type": type_name, "optional": True} def build_inertia_shared_props( app: "Litestar", *, openapi_schema: dict[str, Any] | None, include_default_auth: bool, inertia_config: "InertiaConfig | None", types_config: "TypeGenConfig | None", openapi_support: OpenAPISupport | None = None, ) -> dict[str, dict[str, Any]]: """Build shared props metadata (built-ins + configured props). Args: app: Litestar application instance. openapi_schema: Optional OpenAPI schema dict. include_default_auth: Include default auth shared prop. inertia_config: Optional Inertia configuration. types_config: Optional type generation configuration. openapi_support: Optional shared OpenAPISupport instance. If not provided, a new one will be created. Sharing improves determinism and performance. Returns: Mapping of shared prop name to metadata payload. """ fallback_ts_type = get_fallback_ts_type(types_config) # flash and errors are page-object concerns, not props. The runtime sends flash at # the page top level to match `Page.flash` (inertia/response.py), and @inertiajs/core # declares `Page.props.errors` itself as `Errors & ErrorBag`. Generating either here # promised a `props.flash` that is never present and an error value type # (`string[]`) that disagrees with the `str` values the runtime actually sends. shared_props: dict[str, dict[str, Any]] = {"csrf_token": {"type": "string", "optional": True}} if include_default_auth: shared_props["auth"] = {"type": "AuthData", "optional": True} if inertia_config is None: return shared_props if openapi_support is None: openapi_support = OpenAPISupport.from_app(app, openapi_schema) shared_schema_keys: dict[str, tuple[str, ...]] = {} for key, value in inertia_config.extra_static_page_props.items(): if not key: continue shared_props[key] = {"type": ts_type_from_value(value, fallback_ts_type=fallback_ts_type), "optional": True} if openapi_support.enabled and isinstance(openapi_schema, dict) and should_register_value_schema(value): try: field_def = FieldDefinition.from_annotation(value.__class__) schema_result = openapi_support.schema_creator.for_field_definition(field_def) # type: ignore[union-attr] if isinstance(schema_result, Reference): shared_schema_keys[key] = _get_normalized_schema_key(field_def) except (AttributeError, TypeError, ValueError): # pragma: no cover - defensive pass # Handle session props - can be set[str] or dict[str, type] process_session_props( inertia_config.extra_session_page_props, shared_props, shared_schema_keys, openapi_support, fallback_ts_type ) shared_schema_dicts: dict[str, dict[str, Any]] = {} if inertia_config.shared_page_prop_types: process_shared_prop_types( inertia_config.shared_page_prop_types, shared_props, shared_schema_keys, shared_schema_dicts, openapi_support, fallback_ts_type, ) if not ( openapi_support.context and openapi_support.schema_creator and isinstance(openapi_schema, dict) and (shared_schema_keys or shared_schema_dicts) ): return shared_props generated_components = openapi_support.context.schema_registry.generate_components_schemas() name_map = build_schema_name_map(openapi_support.context.schema_registry) merge_generated_components_into_openapi(openapi_schema, generated_components) for prop_name, schema_key in shared_schema_keys.items(): type_name = name_map.get(schema_key) if type_name: shared_props[prop_name]["type"] = type_name # Containers and unions render from their inline schema once nested $refs resolve. openapi_schema_names = set(openapi_components_schemas(openapi_schema)) | set(generated_components) for prop_name, schema_dict in shared_schema_dicts.items(): shared_props[prop_name]["type"] = normalize_type_string(ts_type_from_openapi(schema_dict), openapi_schema_names) return shared_props
[docs] def generate_inertia_pages_json( app: "Litestar", *, openapi_schema: dict[str, Any] | None = None, include_default_auth: bool = True, include_default_flash: bool = True, inertia_config: "InertiaConfig | None" = None, types_config: "TypeGenConfig | None" = None, ) -> dict[str, Any]: """Generate Inertia pages metadata JSON. The output is deterministic: all dict keys are sorted alphabetically to produce byte-identical output for the same input data. A single OpenAPISupport instance is shared across both page extraction and shared props building to ensure consistent schema registration and naming. This eliminates non-determinism from split schema registries. Returns: An Inertia pages metadata payload as a dictionary with sorted keys. """ # Create a single OpenAPISupport instance to share across the entire pipeline. # This ensures consistent schema registration and prevents "split-brain" issues # where separate registries could produce different schema names. openapi_support = OpenAPISupport.from_app(app, openapi_schema) pages_metadata = extract_inertia_pages( app, openapi_schema=openapi_schema, fallback_type=types_config.fallback_type if types_config is not None else "unknown", openapi_support=openapi_support, ) pages_dict: dict[str, dict[str, Any]] = {} for page in pages_metadata: page_data: dict[str, Any] = {"route": page.route_path} if page.props_type: page_data["propsType"] = page.props_type if page.ts_type: page_data["tsType"] = page.ts_type if page.custom_types: page_data["customTypes"] = page.custom_types if page.wrap_with_content: page_data["wrapWithContent"] = True if page.schema_ref: page_data["schemaRef"] = page.schema_ref if page.handler_name: page_data["handler"] = page.handler_name pages_dict[page.component] = page_data shared_props = build_inertia_shared_props( app, openapi_schema=openapi_schema, include_default_auth=include_default_auth, inertia_config=inertia_config, types_config=types_config, openapi_support=openapi_support, ) # Sort all dict keys for deterministic output # Pages sorted by component name, shared props sorted by prop name sorted_pages = dict(sorted(pages_dict.items())) sorted_shared_props = dict(sorted(shared_props.items())) root: dict[str, Any] = { "fallbackType": types_config.fallback_type if types_config is not None else None, "pages": sorted_pages, "sharedProps": sorted_shared_props, "typeGenConfig": {"includeDefaultAuth": include_default_auth, "includeDefaultFlash": include_default_flash}, "typeImportPaths": types_config.type_import_paths if types_config is not None else None, } # Remove None values for cleaner output return {k: v for k, v in root.items() if v is not None}