Source code for pytest_routes.discovery.litestar
"""Litestar route extraction."""
from __future__ import annotations
import inspect
from typing import Any, get_origin, get_type_hints
from pytest_routes.discovery.base import RouteExtractor, RouteInfo
[docs]
class LitestarExtractor(RouteExtractor):
"""Extract routes from Litestar applications.
This extractor provides first-class support for Litestar framework route discovery.
It extracts comprehensive route metadata including path parameters, query parameters,
request body types, and route metadata from Litestar applications.
The extractor handles both the old and new Litestar API versions, supporting:
- Path parameter extraction with type annotations
- Query parameter detection from handler signatures
- Request body type extraction from DTOs and type hints
- Route metadata (tags, deprecation status, descriptions)
Example:
>>> from litestar import Litestar, get
>>> from pytest_routes.discovery.litestar import LitestarExtractor
>>>
>>> @get("/users/{user_id:int}")
>>> async def get_user(user_id: int, include_posts: bool = False) -> dict:
... return {"id": user_id, "include_posts": include_posts}
>>>
>>> app = Litestar(route_handlers=[get_user])
>>> extractor = LitestarExtractor()
>>> routes = extractor.extract_routes(app)
>>> route = routes[0]
>>> route.path
'/users/{user_id:int}'
>>> route.path_params
{'user_id': <class 'int'>}
>>> route.query_params
{'include_posts': <class 'bool'>}
Note:
The extractor automatically skips HEAD methods as they are typically
auto-generated by the framework and don't need explicit testing.
"""
[docs]
def supports(self, app: Any) -> bool:
"""Check if the application is a Litestar instance.
Args:
app: The ASGI application to check.
Returns:
True if the app is a Litestar instance, False otherwise.
Note:
Returns False if Litestar is not installed, allowing graceful
degradation when the framework is not available.
"""
try:
from litestar import Litestar
return isinstance(app, Litestar)
except ImportError:
return False
[docs]
def extract_routes(self, app: Any) -> list[RouteInfo]:
"""Extract all HTTP and WebSocket routes from a Litestar application.
This method traverses the Litestar route registry and extracts comprehensive
metadata for each route including path parameters, query parameters, request
body types, and route metadata. It handles both HTTP and WebSocket routes.
Args:
app: A Litestar application instance.
Returns:
A list of RouteInfo objects containing complete route metadata:
path (route path pattern), methods (HTTP methods), name (handler name),
handler (function reference), path_params (parameter name to type mapping),
query_params (query parameter mapping), body_type (request body type),
tags (route tags), deprecated (deprecation flag), description (route docs),
is_websocket (True for WebSocket routes), websocket_metadata (WebSocket config).
Example:
>>> from litestar import Litestar, get, post, websocket
>>> from litestar.dto import DTOData
>>> from dataclasses import dataclass
>>>
>>> @dataclass
>>> class CreateUser:
... name: str
... email: str
>>>
>>> @get("/users/{user_id:int}")
>>> async def get_user(user_id: int) -> dict:
... return {"id": user_id}
>>>
>>> @post("/users", tags=["users"])
>>> async def create_user(data: CreateUser) -> dict:
... return {"name": data.name}
>>>
>>> @websocket("/ws/chat")
>>> async def chat_handler(socket) -> None:
... await socket.accept()
... await socket.send_json({"type": "welcome"})
>>>
>>> app = Litestar(route_handlers=[get_user, create_user, chat_handler])
>>> extractor = LitestarExtractor()
>>> routes = extractor.extract_routes(app)
>>> len(routes)
3
>>> routes[0].path_params
{'user_id': <class 'int'>}
>>> routes[1].tags
['users']
>>> routes[2].is_websocket
True
Note:
- HEAD methods are automatically skipped for HTTP routes
- Handles both old API (handler directly) and new API (handler tuple)
- Type hints are extracted from handler functions when available
- Dependencies and framework injection parameters are filtered out
- WebSocket routes have auto_accept=True in their metadata
"""
from litestar.routes import HTTPRoute, WebSocketRoute
routes: list[RouteInfo] = []
for route in app.routes:
if isinstance(route, HTTPRoute):
for method, handler_info in route.route_handler_map.items():
if method == "HEAD":
continue
# Handle both old API (handler directly) and new API (tuple of handler, kwargs_model)
handler = handler_info[0] if isinstance(handler_info, tuple) else handler_info
# Extract type hints from handler
try:
hints = get_type_hints(handler.fn) if handler.fn else {}
except Exception:
hints = {}
path_params = self._extract_path_params(route, hints)
routes.append(
RouteInfo(
path=route.path,
methods=[method],
name=getattr(handler, "name", None),
handler=getattr(handler, "fn", None),
path_params=path_params,
query_params=self._extract_query_params(handler, hints, path_params),
body_type=self._extract_body_type(handler, hints),
tags=list(getattr(handler, "tags", None) or []),
deprecated=getattr(handler, "deprecated", False) or False,
description=getattr(handler, "description", None),
)
)
elif isinstance(route, WebSocketRoute):
routes.append(self._build_websocket_route_info(route))
return routes
def _extract_path_params(self, route: Any, hints: dict[str, Any]) -> dict[str, type]:
"""Extract path parameters and their types."""
params: dict[str, type] = {}
path_parameters = route.path_parameters
# Handle both dict format (new API) and list format (old API)
if isinstance(path_parameters, dict):
for param_name, param_def in path_parameters.items():
# New API: dict with PathParameterDefinition values
if hasattr(param_def, "type"):
params[param_name] = param_def.type
else:
params[param_name] = hints.get(param_name, str)
else:
# Old API: list of dicts with "name" key
for param in path_parameters:
param_name = param["name"] if isinstance(param, dict) else param
params[param_name] = hints.get(param_name, str)
return params
def _extract_query_params( # noqa: C901, PLR0912
self, handler: Any, hints: dict[str, Any], path_params: dict[str, type]
) -> dict[str, type]:
"""Extract query parameters from handler signature.
Query parameters are function parameters that:
- Are not in path_params
- Are not the 'data' parameter (request body)
- Are not 'self' (for controller methods)
- Are not dependencies or state parameters
Args:
handler: The Litestar route handler
hints: Type hints from the handler function
path_params: Already extracted path parameters
Returns:
Dictionary mapping query param names to their types
"""
if not handler.fn:
return {}
query_params: dict[str, type] = {}
try:
sig = inspect.signature(handler.fn)
except (ValueError, TypeError):
return {}
# Get path parameter names
path_param_names = set(path_params.keys())
# Iterate through function parameters
for param_name, param in sig.parameters.items():
# Skip special parameters
if param_name in ("self", "cls"):
continue
# Skip path parameters
if param_name in path_param_names:
continue
# Skip request body parameter (data)
if param_name == "data":
continue
# Skip Litestar dependency injection parameters
# Check type hint first
param_type = hints.get(param_name, param.annotation)
if param_type != inspect.Parameter.empty:
# Check if this is a Litestar injected type
try:
# Common Litestar injected types to skip
type_name = getattr(param_type, "__name__", str(param_type))
# Skip Request, State, scope, and other ASGI types
if any(name in str(type_name) for name in ("Request", "State", "ASGIConnection", "Scope")):
continue
# Also check for 'scope' parameter name (internal Litestar param)
if param_name == "scope":
continue
except Exception: # noqa: S110
# Ignore errors from getattr or string conversion
pass
# Handle Optional types (Union[X, None])
origin = get_origin(param_type)
if origin is not None:
# For Union types, try to extract the non-None type
import types
if hasattr(types, "UnionType") and isinstance(param_type, types.UnionType):
# Python 3.10+ union syntax (X | None)
args = param_type.__args__
non_none_types = [t for t in args if t is not type(None)]
if non_none_types:
param_type = non_none_types[0]
elif hasattr(param_type, "__args__"):
# typing.Union syntax
args = param_type.__args__
non_none_types = [t for t in args if t is not type(None)]
if non_none_types:
param_type = non_none_types[0]
# Default to str if we still don't have a concrete type
if param_type == inspect.Parameter.empty or not isinstance(param_type, type):
param_type = str
query_params[param_name] = param_type
return query_params
def _extract_body_type(self, handler: Any, hints: dict[str, Any]) -> type | None:
"""Extract request body type if present."""
if hasattr(handler, "data_dto") and handler.data_dto:
return handler.data_dto
return hints.get("data")
def _build_websocket_route_info(self, route: Any) -> RouteInfo:
"""Build RouteInfo for a Litestar WebSocket route.
Args:
route: A Litestar WebSocketRoute instance.
Returns:
RouteInfo configured for WebSocket with appropriate metadata.
"""
from pytest_routes.discovery.base import WebSocketMessageType, WebSocketMetadata
handler = route.route_handler
try:
hints = get_type_hints(handler.fn) if handler.fn else {}
except Exception:
hints = {}
path_params = self._extract_path_params(route, hints)
ws_metadata = WebSocketMetadata(
subprotocols=[],
accepted_message_types=[
WebSocketMessageType.TEXT,
WebSocketMessageType.BINARY,
WebSocketMessageType.JSON,
],
sends_message_types=[
WebSocketMessageType.TEXT,
WebSocketMessageType.BINARY,
WebSocketMessageType.JSON,
],
auto_accept=True,
)
return RouteInfo(
path=route.path,
methods=["WEBSOCKET"],
name=getattr(handler, "name", None),
handler=getattr(handler, "fn", None),
path_params=path_params,
query_params={},
body_type=None,
tags=list(getattr(handler, "tags", None) or []),
deprecated=getattr(handler, "deprecated", False) or False,
description=getattr(handler, "description", None),
is_websocket=True,
websocket_metadata=ws_metadata,
)