API Reference

Protocols

CallHandlerProtocol

Wraps the next step in the pipeline.

Interceptors use this to continue processing to the next interceptor or to the actual handler.

async def handle() -> Any

Continue processing the request.

Returns the result of the next handler in the pipeline.

---

ConfigAccessorProtocol

Provides access to web configuration.

Consumers that need configuration details should depend on this protocol for minimal coupling to the provider.

property web_config() -> Any

Return the web-layer configuration.

Returns

Type	Description
Any	WebConfig instance with web-layer settings.

property provider_config() -> Any

Return the provider-specific configuration.

Returns

Type	Description
Any	WebProviderConfig instance with provider settings.

property debug_routes_auth() -> Any

Return the debug routes authentication handler, if set.

Returns

Type	Description
Any	Callable or None.

property fail_on_route_conflict() -> bool

Whether to raise on duplicate route registration.

Returns

Type	Description
bool	True if duplicate routes should raise RuntimeError.

---

ControllerSourceProtocol

Provides the list of registered controller classes.

Consumers that only need access to controllers should depend on this protocol instead of the full WebProvider.

property controllers() -> list[type]

Return the list of registered controller classes.

Returns

Type	Description
list[type]	A list of controller class types.

---

ExecutionContextProtocol

Provides metadata about the current request being processed.

Used by interceptors and guards to access request information and make decisions about how to process the request/response.

property request() -> Any

The current HTTP request.

property handler() -> Callable[Ellipsis, Any]

The handler function that will be called.

property controller_class() -> type | None

The controller class, if this is a controller method.

property method_name() -> str | None

The name of the method being called.

property route_metadata() -> dict[str, Any]

Metadata about the route.

---

PipeProtocol

Transforms or validates a single handler parameter.

Pipes operate on individual parameters before they reach the handler. They can transform (e.g., parse string to int) or validate (e.g., check that a value is within range).

Example

class ParseIntPipe:
    async def transform(self, value: Any, metadata: ParamMetadata) -> int:
        if value is None:
            return metadata.default or 0
        try:
            return int(value)
        except ValueError:
            raise BadRequestError(f"Invalid integer for {metadata.name}")

async def transform(
    value: Any,
    metadata: ParamMetadata
) -> Any

Transform or validate the value.

Parameters

Parameter	Type	Description
`value`	Any	The value to transform/validate.
`metadata`	ParamMetadata	Metadata about the parameter.

Returns

Type	Description
Any	The transformed value.

Raises

Exception	Description
Exception	On validation failure (typically HTTP 400 Bad Request).

---

ProviderResourcesProtocol

Provides access to internal provider resources.

Used by advanced components like routing and middleware managers that need direct access to router and OpenAPI generator.

property router() -> Any

Return the internal Router instance.

Returns

Type	Description
Any	Router instance or None if not initialized.

property openapi_generator() -> Any

Return the OpenAPI generator instance.

Returns

Type	Description
Any	OpenAPIGenerator instance or None if not configured.

---

WebAppAccessorProtocol

Provides access to the underlying Starlette application.

Consumers that only need the Starlette app should depend on this protocol instead of the full WebProvider, decoupling from the provider’s full interface.

property starlette() -> Any

Return the Starlette application instance.

Returns

Type	Description
Any	The Starlette ASGI application, or None if not yet initialized.

---

WebInterceptorProtocol

Intercepts request/response flow with full AOP control.

Interceptors wrap the entire request→handler→response lifecycle, enabling cross-cutting concerns like logging, caching, response transformation, and timing without modifying handler code.

Example

class TimingInterceptor(Interceptor):
    async def intercept(
        self, context: ExecutionContextProtocol, next: CallHandlerProtocol
    ) -> Any:
        start = time.perf_counter()
        result = await next.handle()
        elapsed = time.perf_counter() - start
        return result

async def intercept(
    context: ExecutionContextProtocol,
    next_handler: CallHandlerProtocol
) -> Any

Intercept the request/response flow.

Parameters

Parameter	Type	Description
`context`	ExecutionContextProtocol	Provides metadata about the current request.
`next_handler`	CallHandlerProtocol	Wraps the next step in the pipeline.

Returns

Type	Description
Any	The result of the handler (possibly transformed).

Notes: - MUST call await next_handler.handle() to continue the pipeline. - CAN transform the result before returning. - CAN add/modify response headers. - CAN short-circuit by returning early without calling next.

---

WebProviderProtocol

Combined protocol for full WebProvider access.

This is the complete interface expected by components that need the full capabilities of WebProvider. Consumers that need fewer properties should depend on the more specific protocols above to reduce coupling.

---

Classes

ApiVersionMetadata

Metadata attached to a versioned controller or handler by ``@api_version``.

property url_prefix() -> str

Return the URL prefix for this version.

---

BackgroundTasks

Accumulates background tasks to run after response is sent.

Context is captured at add time using propagate_context so that _execute_all restores the request-scope context variables (request_id, trace_id, …) even when called after the original request scope has been torn down.

def __init__() -> None

def add(
    func: Callable[Ellipsis, Any],
    *args: Any,
    **kwargs: Any
) -> None

Add a task to be executed after the response.

The current contextvars context is snapshotted immediately so that _execute_all can restore it at execution time.

---

CQRSController

Controller with built-in CQRS command and query bus dispatch.

Subclass this instead of Controller when your endpoints need to dispatch commands or queries through the CQRS buses.

The buses are injected via the constructor and are expected to be resolved from the DI container.

Example

class OrderController(CQRSController):
    @post("/orders")
    async def create_order(self, request: Request) -> JSONResponse:
        data = await request.json()
        result = await self.dispatch_command(CreateOrderCommand(**data))
        return JSONResponse({"id": result.id}, status_code=201)

    @get("/orders/{order_id}")
    async def get_order(self, request: Request) -> JSONResponse:
        order_id = request.path_params["order_id"]
        result = await self.dispatch_query(GetOrderQuery(order_id=order_id))
        return JSONResponse(result.to_dict())

def __init__(
    command_bus: CommandBusProtocol | None = None,
    query_bus: QueryBusProtocol | None = None
) -> None

Initialize CQRSController with optional command and query buses.

Parameters

Parameter	Type	Description
`command_bus`	CommandBusProtocol | None	CommandBusProtocol implementation for dispatching commands. If None, calling ``dispatch_command`` will raise RuntimeError.
`query_bus`	QueryBusProtocol | None	QueryBusProtocol implementation for executing queries. If None, calling ``dispatch_query`` will raise RuntimeError.

async def dispatch_command(command: Any) -> Any

Dispatch a command through the command bus.

Parameters

Parameter	Type	Description
`command`	Any	The command object to dispatch.

Returns

Type	Description
Any	The result returned by the command handler.

Raises

Exception	Description
RuntimeError	If no CommandBusProtocol was provided to this controller.

async def dispatch_query(query: Any) -> Any

Dispatch a query through the query bus.

Parameters

Parameter	Type	Description
`query`	Any	The query object to execute.

Returns

Type	Description
Any	The query result.

Raises

Exception	Description
RuntimeError	If no QueryBusProtocol was provided to this controller.

---

Controller

Base controller class with dependency injection support.

Controllers group related route handlers and can specify dependencies to be injected by the DI container.

Example

class UserController(Controller): @get(“/users”) async def list_users(self): return []

def __init__() -> Any

def collect_routes(cls) -> list[dict[str, Any]]

Collect routes from controller methods.

It scans the class and its base classes for methods with _route_config attribute and returns a list of route definitions.

Returns

Type	Description
List of route dictionaries with keys	method, path, handler_name, etc.

---

ControllerRegistry

Registry for managing controller registrations.

Canonical source for controllers: ControllerRegistry. Use ControllerRegistry for controller-level operations (class registration). Use RouteRegistry for route-level operations (path, method, handler).

def __init__() -> None

def register(
    key: str | type,
    value: type | None = None,
    *,
    allow_overwrite: bool | None = None
) -> type | Any

Register a controller class.

Parameters

Parameter	Type	Description
`key`	str | type	Component key or class
`value`	type | None	The controller class to register
`allow_overwrite`	bool | None	Whether to allow overwriting an existing registration

Returns

Type	Description
type | Any	The registered controller class

def get(name: str) -> type | None

Get a controller class by name.

def list_controllers() -> list[str]

List all registered controller names.

def get_all_controllers() -> list[type]

Get all registered controller classes.

def clear() -> None

Clear all registrations.

Use in tests only to prevent test pollution between test cases.

---

DefaultMiddlewareStack

Builds the default Lexigram middleware stack as an explicit, composable list.

Encapsulates the minimum set of middlewares that every Lexigram web application applies so users can see and extend it without magic.

WebProvider delegates to this class internally, ensuring that all code paths produce the same default stack.

Parameters

Parameter	Type	Description
`container`		The resolved DI container. Required for DIScopeMiddleware which provides request-scoped dependency resolution.
`extra_middlewares`		Additional user-supplied Lexigram middlewares to include in the stack. They are adapted to Starlette ``Middleware`` wrappers via MiddlewareAdapterRegistry and inserted before the built-in defaults.

Example

from lexigram.web.middleware import DefaultMiddlewareStack
from lexigram.logging import get_logger

logger = get_logger(__name__)

# Inspect the default stack
stack = DefaultMiddlewareStack(container=container)
for mw in stack.build():
    logger.debug("middleware", name=mw.cls.__name__)

# Extend with custom middleware
stack = DefaultMiddlewareStack(
    container=container,
    extra_middlewares=[MyTimingMiddleware()],
)

def __init__(
    container: ContainerResolverProtocol | None = None,
    extra_middlewares: list[Any] | None = None
) -> None

Initialise the stack builder.

Parameters

Parameter	Type	Description
`container`	ContainerResolverProtocol | None	Resolved DI container for ``DIScopeMiddleware``.
`extra_middlewares`	list[Any] | None	Optional extra Lexigram middlewares to include.

def build() -> list[StarletteMiddleware]

Build and return the ordered list of default Starlette middlewares.

The always-present entry is DIScopeMiddleware. Any extra_middlewares supplied at construction time are prepended to that entry after adaptation.

Returns

Type	Description
list[StarletteMiddleware]	An ordered list of starlette.middleware.Middleware instances ready to pass to starlette.applications.Starlette.

---

Err

def __init__(error: E) -> None

def is_ok() -> bool

def is_err() -> bool

def unwrap() -> T

def unwrap_err() -> E

def unwrap_or(default: T) -> T

def unwrap_or_else(op: Callable[[E], T]) -> T

def map_sync(op: Callable[[T], U]) -> Result[U, E]

def map_err(op: Callable[[E], F]) -> Result[T, F]

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

def expect(message: str) -> T

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

async def async_map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

Alias for map — exists for backward compatibility.

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

def flatten() -> Result[Any, E]

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

def ok_or(default: U) -> U

---

FastJSONResponse

JSON response using lexigram common JSON utilities.

Uses orjson when available (5-10x faster than stdlib json) with graceful fallback to stdlib json. Provides consistent JSON serialization across all Lexigram packages.

Performance benefits:

• 5-10x faster than stdlib json when orjson is available
• Native datetime/UUID/dataclass support
• Efficient bytes output
• Consistent behavior across packages

For Pydantic models, prefer model.model_dump_json() which uses Rust-based serialization (similar performance to orjson).

def render(content: Any) -> bytes

---

FileResponse

File response for serving static files

---

GenericController

Generic controller with common CRUD patterns for any service.

Returns Result[T, DomainError] from all action methods. The ResponseSerializer in the request pipeline automatically maps:

• Ok(value) → 200 JSON response (201 for create_item).
• Err(error) → HTTP error via ResultResponseMapper.

Use @error_status on your domain error classes to control the HTTP status code they map to

from lexigram.web import error_status

@error_status(404)
class ItemNotFound(DomainError): ...

def __init__(
    service: CRUDServiceProtocol[T],
    resource_name: str | None = None
) -> None

property resource_name() -> str

Get the resource name for permissions and error messages.

property logger() -> Any

Get the logger for this controller.

async def list_items(
    limit: int = 20,
    offset: int = 0,
    **filters: Any
) -> Result[dict[str, Any], DomainError]

List items with pagination.

Returns a Result whose Ok payload is a dict with keys items, limit, offset, and total.

async def get_item(item_id: str) -> Result[T, DomainError]

Get single item by ID.

async def create_item(data: dict[str, Any]) -> Any

Create new item (returns 201 on success).

async def update_item(
    item_id: str,
    data: dict[str, Any]
) -> Result[T, DomainError]

Update existing item.

async def delete_item(item_id: str) -> Any

Delete item (returns 204 on success).

---

HTMLContent

Marker type explicitly indicating HTML content.

Handlers that return HTMLContent are explicitly declaring that the returned string should be treated as HTML. This avoids heuristic checks and makes the behavior explicit for callers.

---

HTMLResponse

HTML response

def __init__(
    content: Any = None,
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    media_type: str | None = None,
    background: Any | None = None
) -> None

---

HTMXResponse

Convenience HTML response with HTMX-specific headers support.

Example

return HTMXResponse(“

ok

”, hx_trigger={“showToast”: “Saved”}) This will set the HX-Trigger header to the JSON-encoded value.

def __init__(
    content: Any,
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    hx_trigger: Any = None,
    hx_refresh: bool = False
) -> None

---

InputSanitizationMiddleware

ASGI middleware that sanitizes query string parameters.

Strips null bytes and rejects obvious script-injection patterns from query parameters before they reach request handlers. Path parameters are not modified because they have been parsed and validated by the router before they hit middleware.

This is a defense-in-depth measure, not a substitute for validation at the service layer.

Parameters

Parameter	Type	Description
`app`		The ASGI application to wrap.
`sanitize_query_params`		Whether to sanitize query string values (default: ``True``).

def __init__(
    app: Any,
    sanitize_query_params: bool = True
) -> None

---

JSONResponse

JSON response with high performance.

Uses lexigram common JSON utilities for consistent serialization across all Lexigram packages. Automatically uses Pydantic’s Rust serializer for Pydantic models when available.

def __init__(
    content: Any = None,
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    media_type: str | None = None,
    background: Any | None = None
) -> None

---

Ok

def __init__(value: T) -> None

def is_ok() -> bool

def is_err() -> bool

def unwrap() -> T

def unwrap_err() -> E

def unwrap_or(default: T) -> T

def unwrap_or_else(op: Callable[[E], T]) -> T

def map_sync(op: Callable[[T], U]) -> Result[U, E]

def map_err(op: Callable[[E], F]) -> Result[T, F]

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

def expect(message: str) -> T

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

async def async_map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

Alias for map — exists for backward compatibility.

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

def flatten() -> Result[Any, E]

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

def ok_or(default: U) -> T

---

ParamMetadata

Metadata about a parameter being piped.

Attributes: name: The parameter name. param_type: The type of parameter (path, query, body, header, cookie, file). expected_type: The expected Python type for the parameter. default: Default value if the parameter is not provided. alias: Optional alias for the parameter.

---

ProblemDetail

RFC 7807 Problem Details for HTTP APIs.

Attributes: type: URI identifying the problem type title: Short summary of the problem status: HTTP status code detail: Human-readable explanation instance: URI for this specific occurrence errors: Additional validation errors (extension)

def to_dict() -> dict[str, Any]

Convert to dictionary for JSON serialization.

def from_exception(
    cls,
    exc: Exception,
    status: int = 500,
    debug: bool = False,
    **kwargs
) -> ProblemDetail

Create a ProblemDetail from an exception.

When debug is False (default), server-error responses (5xx) use a generic detail string to prevent internal information from leaking to API consumers. Set debug=True only in development environments.

def validation_error(
    cls,
    errors: list[dict[str, Any]],
    detail: str = 'Validation failed'
) -> ProblemDetail

Create a validation error ProblemDetail.

def not_found(
    cls,
    resource: str,
    identifier: str | None = None
) -> ProblemDetail

Create a not found ProblemDetail.

def bad_request(
    cls,
    detail: str,
    errors: list[dict[str, Any]] | None = None
) -> ProblemDetail

Create a bad request ProblemDetail.

def internal_error(
    cls,
    detail: str = 'An internal error occurred'
) -> ProblemDetail

Create an internal error ProblemDetail.

---

RateLimitModule

Module-level rate limiting configuration.

Registers a RateLimitProvider in the application’s DI container so that route-level @throttle and @rate_limit decorators can resolve a limiter from the container.

Usage

app.add_module(RateLimitModule.configure(
    backend="redis",
    default_limit="100/minute",
))

# Development / testing — no external service required:
app.add_module(RateLimitModule.configure(backend="memory"))

def configure(
    cls,
    *,
    backend: Literal['redis', 'memory'] = 'memory',
    default_limit: str | None = None,
    redis_client: Any = None
) -> Any

Create a DynamicModule that wires up rate limiting.

Parameters

Parameter	Type	Description
`backend`	Literal['redis', 'memory']	Storage backend. ``"redis"`` uses atomic Lua scripts (recommended for production); ``"memory"`` uses a process-local sliding window (suitable for development and single-process deployments).
`default_limit`	str | None	Optional default rate limit applied globally (not yet enforced by the module — provided for future middleware use). Accepts the same rate string format as throttle.
`redis_client`	Any	Pre-built Redis client. Only used when ``backend="redis"``. When *None*, the provider tries to resolve a ``"redis_client"`` binding from the container at boot time.

Returns

Type	Description
Any	A DynamicModule descriptor.

---

RateLimitProvider

Standalone provider for rate-limiting services.

Registers a RateLimiter in the DI container and lazily resolves a Redis client from the container at boot time when none is supplied at construction.

def __init__(
    redis_client: Any = None,
    enabled: bool = True
) -> None

Initialize rate limit provider.

Parameters

Parameter	Type	Description
`redis_client`	Any	Optional pre-built Redis client for rate limiting.
`enabled`	bool	When ``False`` the provider skips all registration/boot.

async def register(container: ContainerRegistrarProtocol) -> None

Register rate limiter in the DI container.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	DI container registrar.

async def boot(container: ContainerResolverProtocol) -> None

Start the rate limit provider, resolving Redis from the container if needed.

async def shutdown() -> None

Shutdown the rate limit provider.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check rate-limit provider health.

Returns

Type	Description
HealthCheckResult	HealthCheckResult reflecting whether the limiter is initialised.

---

RedirectResponse

Redirect response

---

Request

Lexigram request wrapper around Starlette request

def __init__(starlette_request: StarletteRequest)

property method() -> str

property url() -> Any

property headers() -> Any

property query_params() -> dict[str, str]

property cookies() -> dict[str, str]

property client() -> Any

Client address information

property app() -> Any

Forward to the underlying Starlette app

property session() -> dict[str, Any]

Access the session state

property path_params() -> dict[str, Any]

async def json() -> Any

async def body() -> bytes

async def form() -> Any

property state() -> Any

Request state for middleware

property scope() -> dict[str, Any]

Access the underlying ASGI scope

property request() -> StarletteRequest

Access the underlying Starlette request

property user() -> Any | None

Authenticated user set by auth middleware.

property auth() -> Any | None

Authentication credentials.

property request_id() -> str | None

Request ID set by RequestIDMiddleware.

property is_json() -> bool

Check if request content type is JSON.

property accepted_types() -> list[str]

Parse Accept header into ordered list.

property ip() -> str

Client IP address (proxy-aware).

property validated_data() -> dict[str, Any]

Data validated/transformed by pipe pipeline.

def validated_data(value: dict[str, Any]) -> None

Set validated data from pipe pipeline.

---

Response

Base response class

---

Result

Base Result type. Not abstract — Ok and Err are the only variants.

def is_ok() -> bool

def is_err() -> bool

def unwrap() -> T

def unwrap_err() -> E

def unwrap_or(default: T) -> T

def unwrap_or_else(op: Callable[[E], T]) -> T

def map_sync(op: Callable[[T], U]) -> Result[U, E]

def map_err(op: Callable[[E], F]) -> Result[T, F]

def and_then_sync(op: Callable[[T], Result[U, E]]) -> Result[U, E]

def or_else_sync(op: Callable[[E], Result[T, F]]) -> Result[T, F]

def expect(message: str) -> T

def match(
    ok: Callable[[T], U],
    err: Callable[[E], U]
) -> U

async def map(op: Callable[[T], Awaitable[U]]) -> Result[U, E]

async def and_then(op: Callable[[T], Awaitable[Result[U, E]]]) -> Result[U, E]

async def or_else(op: Callable[[E], Awaitable[Result[T, F]]]) -> Result[T, F]

def flatten() -> Result[Any, E]

def filter(
    predicate: Callable[[T], bool],
    error: E
) -> Result[T, E]

def ok_or(default: U) -> T | U

def from_exception(
    cls,
    exc: Exception,
    ok_type: type[T] = type(None)
) -> Result[T, Exception]

Wrap a caught exception into an Err result.

def to_optional() -> T | None

def inspect(op: Callable[[T], None]) -> Result[T, E]

def inspect_err(op: Callable[[E], None]) -> Result[T, E]

---

ResultResponseMapper

Maps ``Result`` error values to HTTP responses.

The default mappings follow REST conventions:

Error type	Status
NotFoundError	404
ValidationError	422
PermissionDeniedError	403
AuthorizationError	403
AuthenticationError	401
ConflictError	409
RateLimitError	429
DomainError (base)	400
other	400

def to_response(
    result: Any,
    success_status: int = 200
) -> Response

Convert a Result to an HTTP response.

Parameters

Parameter	Type	Description
`result`	Any	The ``Result[T, E]`` object to map.
`success_status`	int	HTTP status code for Ok results (default 200).

Returns

Type	Description
Response	A JSONResponse with appropriate status code and structured body.

def register(
    cls,
    error_type: type[Exception],
    status_code: int
) -> None

Register a custom error type → HTTP status mapping.

Custom registrations take precedence over defaults (inserted at front).

Parameters

Parameter	Type	Description
`error_type`	type[Exception]	The exception class to match.
`status_code`	int	HTTP status code to return.

def error_to_response(
    cls,
    error: Any
) -> Response

Convert a domain error to a JSON HTTP response.

Parameters

Parameter	Type	Description
`error`	Any	The error value from ``Result.unwrap_err()``.

Returns

Type	Description
Response	A JSONResponse with the appropriate status code and a structured body.

---

RoleGuard

GuardProtocol that checks if user has required role(s).

The authorizer must be injected at guard instantiation time (constructor injection).

Usage

authorizer = await container.resolve(AuthorizerProtocol) @use_guards(RoleGuard(*required_roles, authorizer=authorizer)) async def admin_only(self): …

def __init__(
    *required_roles: str,
    authorizer: AuthorizerProtocol
) -> None

async def can_activate(context: Any) -> bool

Check if user has required role using AuthorizerProtocol.

---

RouteRegistry

Registry for managing route registrations and discovery.

Canonical source for routes: RouteRegistry. Use RouteRegistry for route-level operations (path, method, handler). Use ControllerRegistry for controller-level operations (class registration).

def __init__(debug: bool = False)

def register_controller(
    controller_cls: type[Controller],
    controller_registry: ControllerRegistry | None = None
) -> None

Register a controller class and synchronize to ControllerRegistry.

Parameters

Parameter	Type	Description
`controller_cls`	type[Controller]	The controller class to register.
`controller_registry`	ControllerRegistry | None	If provided, also registers the class by name in the ControllerRegistry for atomic synchronization. This ensures a controller exists in both registries, preventing lookup inconsistencies.

def get_all_routes() -> dict[str, dict[str, Any]]

Get all registered routes

def get_controllers() -> list[type[Controller]]

Get all registered controllers

def find_route(
    path: str,
    method: str
) -> dict[str, Any] | None

Find a route by path and method

def get_routes_by_controller(controller_cls: type[Controller]) -> list[dict[str, Any]]

Get all routes for a specific controller

def clear() -> None

Clear all registrations

---

Router

Router with Controller DI Strategy support.

The Router manages route registration, controller resolution, and request handling. It integrates with the DI container to provide automatic dependency injection for controllers and their methods.

Attributes: routes: List of registered routes with their handlers. _controller_cache: Cache of resolved controller instances. _container: DI container for resolving dependencies. _signature_cache: Cache of handler signatures for performance. _routes_by_name: Dictionary of routes indexed by their unique name. exception_filters: Registry for handling controller exceptions.

def __init__(filter_pipeline: FilterPipeline | None = None)

def get_route_by_name(name: str) -> Route | None

Get a route by its unique name in O(1) time.

Parameters

Parameter	Type	Description
`name`	str	The unique name of the route.

Returns

Type	Description
Route | None	The Route object if found, otherwise None.

def clear_cache() -> None

Clear all cached handler signatures and type-hint lookups.

Call this when modules are hot-reloaded so that stale signatures are not used. The caches will be repopulated lazily on the next request that reaches each handler.

Example

hot_reload_manager.on_reload(router.clear_cache)

def add_route(
    method: str,
    path: str,
    handler: Callable,
    name: str | None = None,
    controller_cls: type | None = None,
    **kwargs: Any
) -> None

Add a route (used internally)

---

StreamingResponse

Streaming response for large files or server-sent events

---

VersionExtractor

Extracts version from requests based on strategy.

def __init__(config: VersioningConfig)

def extract(request: Request) -> str

Extract version from request.

---

WebConfig

Hierarchical root configuration for Lexigram Web.

This is the single source of truth for web configuration, loaded from application.yaml’s web section.

Attributes: name: Configuration name (default: “web”) enabled: Whether the web module is enabled server: Server binding and worker settings app: Application-level settings (OpenAPI, CORS origins, etc.) security: Security headers and policies cors: CORS configuration rate_limit: Rate limiting rules debug_routes: Enable /debug/* endpoints enable_identity_resolution: Resolve OAuth external IDs to internal UUIDs enable_auth: Enable built-in authentication middleware auth_exclude_paths: Paths excluded from authentication

def validate_production_security() -> WebConfig

Block insecure configurations in production.

---

WebInterceptorBase

Base class for interceptors (optional convenience).

Provides a no-op implementation that subclasses can override.

async def intercept(
    context: ExecutionContextProtocol,
    next_handler: CallHandlerProtocol
) -> Any

Default implementation just passes through.

---

WebModule

Web module with HTTP provider.

def configure(
    cls,
    controllers: list[type] | None = None,
    discover: list[str] | tuple[str, Ellipsis] | None = None,
    host: str | None = None,
    port: int | None = None,
    **kwargs: Any
) -> DynamicModule

Create a WebModule with explicit configuration.

Configuration (CORS, CSRF, server host/port, etc.) is loaded from the [web] section of application.yaml by the orchestrator and injected into the provider at registration time. Pass a web_config kwarg only if you need to bypass YAML entirely.

Parameters

Parameter	Type	Description
`controllers`	list[type] | None	Controller classes to register with the web server.
`discover`	list[str] | tuple[str, Ellipsis] | None	Package paths to scan for controllers and merge with any explicitly provided controller classes.
`host`	str | None	Override the server host (builds a ``WebConfig`` internally).
`port`	int | None	Override the server port (builds a ``WebConfig`` internally). **kwargs: Additional keyword arguments forwarded to WebProvider.

def stub(cls) -> DynamicModule

Return a no-op WebModule suitable for unit testing.

Registers WebProvider with no controllers and test-safe configuration. No real HTTP server is started.

Returns

Type	Description
DynamicModule	A DynamicModule with in-memory web configuration.

---

WebProvider

Web provider with native Starlette integration and pass-through architecture.

Recommended usage — let the orchestrator inject configuration via application.yaml (uses config_key = "web" and config_model = WebConfig)

app.add_provider(WebProvider())

Config-first factory — construct from an explicit WebConfig

from lexigram.web import WebProvider, WebConfig

app.add_provider(WebProvider.from_config(WebConfig(debug=True, ...)))

Advanced — pass individual components explicitly (e.g. for tests)

app.add_provider(WebProvider(
    controllers=[UsersController],
    middleware=[AuthMiddleware],
    web_config=WebConfig(debug=True),
))

Note: The multi-parameter constructor is intended for advanced and test scenarios. For typical applications, prefer the no-arg or from_config() form so that configuration is driven by application.yaml.

def __init__(
    middleware: list[Any] | None = None,
    exception_handlers: dict[Any, Any] | None = None,
    controllers: list[type[Controller]] | None = None,
    web_config: WebConfig | None = None,
    provider_config: WebProviderConfig | None = None,
    debug_routes_auth: Callable[Ellipsis, Any] | None = None
) -> None

property contributor_registry() -> Any

Expose contributor registry for route setup access.

def from_config(
    cls,
    config: WebConfig,
    **context: Any
) -> WebProvider

Create a WebProvider from config.

Context kwargs may include middleware, controllers, exception_handlers.

def auto_discover(
    cls,
    *packages: str,
    web_config: WebConfig | None = None,
    **kwargs: Any
) -> WebProvider

Create a WebProvider with controllers auto-discovered from packages.

Scans each package recursively for Controller subclasses and registers them automatically.

Parameters

Parameter	Type	Description
`web_config`	WebConfig | None	Optional web configuration. Falls back to defaults. **kwargs: Extra kwargs forwarded to WebProvider.__init__.

Returns

Type	Description
WebProvider	A configured WebProvider instance.

Example

app.add_provider(WebProvider.auto_discover("my_app.api.controllers"))

async def register(container: ContainerRegistrarProtocol) -> None

Register web services in DI container

async def boot(container: ContainerResolverProtocol) -> None

Initialize the web layer in five ordered phases.

Phase 1 — OpenAPI generator Instantiates OpenAPIGenerator with title and version from web_config.

Phase 2 — Starlette application Builds the native ASGI Starlette instance. The middleware stack is composed once here; it is never rebuilt at request time. Container and config are attached to app.state.

Phase 3 — Middleware pipeline Iterates registered AbstractMiddleware subclasses and wraps the Starlette app. Order follows the provider registration order (outermost-first).

Phase 4 — Integration setup Wires optional first-class integrations: authentication, rate limiting, GraphQL gateway. Each integration is only activated when its config key is present in the resolved container.

Phase 5 — Route registration Discovers annotated controller methods and mounts them on the Starlette router. Route-level dependencies (guards, interceptors, serializers) are resolved here.

async def shutdown() -> None

Cleanup resources created by this provider.

async def health_check(timeout: float = 5.0) -> HealthCheckResult

Check web provider health.

def run_server(
    host: str = '127.0.0.1',
    port: int = 8000,
    **kwargs: Any
) -> None

Run the web application using Granian.

See lexigram.web.server.runner.run_server for implementation.

---

WebRequestReceivedHook

Payload fired when the ASGI server receives an incoming HTTP request.

Attributes: path: URL path of the incoming request (e.g. "/api/users"). method: HTTP verb in upper-case (e.g. "GET").

---

WebResponsePreparedHook

Payload fired once a response has been assembled, before it is sent.

Attributes: path: URL path the response belongs to. status_code: HTTP status code of the prepared response.

---

WebServerStartedHook

Payload fired after the ASGI lifespan ``startup`` phase completes.

---

WebServerStoppedHook

Payload fired after an orderly ASGI lifespan ``shutdown`` completes.

---

WebSocket

Lexigram WebSocket wrapper around Starlette WebSocket

def __init__(starlette_ws: StarletteWebSocket)

async def accept(subprotocol: str | None = None) -> None

async def close(code: int = 1000) -> None

async def receive_text() -> str

async def receive_json() -> Any

async def send_text(data: str) -> None

async def send_json(data: Any) -> None

async def send_bytes(data: bytes) -> None

property state() -> Any

WebSocket state for middleware

---

Functions

api_version

def api_version(
    ver: int | str,
    *,
    deprecated: bool = False,
    sunset: str | None = None,
    prefix: str | None = None
) -> Callable[[Any], Any]

Mark a controller (or individual handler) with an API version.

This extends the simpler version decorator with:

• URL prefix — the version number is automatically prepended to the controller’s prefix attribute (e.g. prefix = "/users" becomes "/v1/users"). Override with the prefix argument.
• Deprecation support — setting deprecated=True causes the routing layer to inject Deprecation: true and optionally Sunset: <date> response headers for all routes on the controller.
• Metadata storage — an ApiVersionMetadata instance is stored on the class/function as __api_version_meta__ and the plain version string as __api_version__ (compatible with VersioningMiddleware).

Parameters

Parameter	Type	Description
`ver`	int | str	Version number or string (e.g. ``1``, ``"2"``, ``"2.1"``).
`deprecated`	bool	When ``True``, mark this version as deprecated.
`sunset`	str | None	Optional ISO-8601 date string indicating when support ends.
`prefix`	str | None	Explicit URL prefix to use instead of the auto-derived one.

Returns

Type	Description
Callable[[Any], Any]	Class/function decorator.

Example

@api_version(1)
class UserControllerV1(Controller):
    prefix = "/users"   # mounted at /v1/users

@api_version(2)
class UserControllerV2(Controller):
    prefix = "/users"   # mounted at /v2/users

@api_version(1, deprecated=True, sunset="2025-12-31")
class LegacyController(Controller):
    prefix = "/legacy"  # adds Deprecation + Sunset headers

---

background

def background(func: Callable[Ellipsis, Awaitable[Any]]) -> Callable[Ellipsis, Awaitable[Any]]

Decorator to run a function as a background task.

---

body

def body(
    default: Any = ...,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

Request body parameter decorator.

---

controller

def controller(name: str | None = None) -> Any

Decorator to automatically register a controller class.

Parameters

Parameter	Type	Description
`name`	str | None	Optional controller name, defaults to class name

Example

@controller() class UserController(Controller): @get(“/users”) async def get_users(self): return {“users”: []}

---

cookie

def cookie(
    default: Any = ...,
    alias: str | None = None,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

Cookie parameter decorator.

---

delete

def delete(
    path: str,
    **kwargs: Any
) -> Any

Register a DELETE route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

---

discover_controllers

def discover_controllers(packages: tuple[str, Ellipsis] | list[str]) -> list[type]

Scan packages for Controller subclasses and return the classes.

Parameters

Parameter	Type	Description
`packages`	tuple[str, Ellipsis] | list[str]	Dotted Python package paths to scan.

Returns

Type	Description
list[type]	List of Controller subclasses (not instantiated — the DI container constructs them during boot).

---

error_status

def error_status(
    error_type: type[Exception],
    status_code: int
) -> Any

Decorator to register a custom error → HTTP status mapping.

Parameters

Parameter	Type	Description
`error_type`	type[Exception]	Exception class to match.
`status_code`	int	HTTP status code to use.

Example

@error_status(PaymentDeclined, 402)
class BillingController(Controller):
    ...

---

file

def file(
    default: Any = ...,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

File upload parameter decorator.

---

file_response

def file_response(
    path: str | Path,
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    media_type: str | None = None,
    filename: str | None = None,
    content_disposition_type: str = 'attachment'
) -> FileResponse

Create a file response

---

form

def form(
    default: Any = ...,
    alias: str | None = None,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

Form parameter decorator.

---

get

def get(
    path: str,
    **kwargs: Any
) -> Any

Register a GET route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

Example

@get(“/users”) async def list_users(request): … return {“users”: []}

---

get_current_user_optional

async def get_current_user_optional(request: Request) -> Any | None

Get current user from state (optional).

---

get_current_user_required

async def get_current_user_required(request: Request) -> Any

Get current user from state (required).

---

get_request_id

async def get_request_id(request: Request) -> str

Get request ID from state.

---

get_version

def get_version(request: Request) -> str

Get API version from request state.

Parameters

Parameter	Type	Description
`request`	Request	The HTTP request

Returns

Type	Description
str	API version string

---

guard

def guard(*guard_classes: type[Any] | Any) -> Any

Concise alias for use_guards.

Attaches one or more guards to a route handler or controller class. Guards are executed in order; the first failure returns a 403 response.

Parameters

Parameter	Type	Description

Returns

Type	Description
Any	Decorator that attaches guards to the target.

Example

@guard(AuthGuard)
async def profile(self): ...

@guard(AuthGuard, PermissionGuard("users:write"))
async def create_user(self): ...

---

head

def head(
    path: str,
    **kwargs: Any
) -> Any

Register a HEAD route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

---

header

def header(
    default: Any = ...,
    alias: str | None = None,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

Header parameter decorator.

---

html_response

def html_response(
    content: str,
    status_code: int = 200,
    headers: dict[str, str] | None = None
) -> HTMLResponse

Create an HTML response

---

injectable

def injectable(cls: type[T]) -> type[T]

Mark cls as transient and auto-register it in the quickstart container.

Each DI resolution creates a new instance of the class (transient scope). This is the quickstart alias for the transient-scope decorator.

Returns the class unchanged, so the decorator is transparent.

Parameters

Parameter	Type	Description
`cls`	type[T]	The class to register as a transient (per-resolve) service.

Returns

Type	Description
type[T]	The original *cls* with the injectable marker applied.

Example

from lexigram.web import app, get, injectable
from lexigram.logging import get_logger

logger = get_logger(__name__)

@injectable
class RequestLogger:
    def log(self, msg: str) -> None:
        logger.info("request_log", message=msg)

---

json_response

def json_response(
    content: Any,
    status_code: int = 200,
    headers: dict[str, str] | None = None
) -> JSONResponse

Create a JSON response

---

options

def options(
    path: str,
    **kwargs: Any
) -> Any

Register an OPTIONS route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

---

patch

def patch(
    path: str,
    **kwargs: Any
) -> Any

Register a PATCH route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

---

path

def path(
    alias: str | None = None,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

Path parameter decorator.

---

post

def post(
    path: str,
    **kwargs: Any
) -> Any

Register a POST route handler.

Parameters annotated with a Pydantic BaseModel or DomainModel subclass are automatically deserialised from the JSON request body by ParameterBinder — no body() decorator is required.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

Example

@post(“/users”) async def create_user(request): … data = await request.json() … return {“id”: 1, “data”: data}

---

put

def put(
    path: str,
    **kwargs: Any
) -> Any

Register a PUT route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

---

query

def query(
    default: Any = ...,
    alias: str | None = None,
    validation: Callable[Ellipsis, Any] | None = None,
    pipes: list[Any] | None = None
) -> Any

Query parameter decorator.

---

redirect_response

def redirect_response(
    url: str,
    status_code: int = 302,
    headers: dict[str, str] | None = None
) -> RedirectResponse

Create a redirect response

---

register_controller

def register_controller(controller_cls: type[Controller]) -> None

Convenience function to register a controller.

Registers the controller in both RouteRegistry and ControllerRegistry, ensuring synchronization and preventing lookup inconsistencies.

---

render_template

def render_template(
    name: str,
    context: dict[str, Any] | None = None,
    templates: Jinja2Templates | None = None,
    **kwargs
) -> str

---

roles

def roles(
    *role_names: str,
    authorizer: Any | None = None
) -> Any

Restrict a handler to users that have at least one of the given roles.

The authorizer dependency must be injected.

Parameters

Parameter	Type	Description
`authorizer`	Any | None	**Required** AuthorizerProtocol instance, typically resolved from the container during application startup.

Returns

Type	Description
Any	Decorator that attaches a RoleGuard to the target.

Example

authorizer = await container.resolve(AuthorizerProtocol)

@roles("admin", authorizer=authorizer)
async def admin_only(self): ...

@roles("admin", "moderator", authorizer=authorizer)
async def manage_content(self): ...

---

singleton

def singleton(cls: type[T]) -> type[T]

Mark cls as a singleton and auto-register it in the quickstart container.

Combines the lexigram.di.singleton DI marker (which sets __lexigram_injectable__) with a direct entry in the quickstart service registry so the class is discovered regardless of whether it is defined at module scope.

Returns the class unchanged, so the decorator is transparent.

Parameters

Parameter	Type	Description
`cls`	type[T]	The class to register as a container-managed singleton.

Returns

Type	Description
type[T]	The original *cls* with the injectable marker applied.

Example

from lexigram.web import app, get, singleton

@singleton
class UserRepo:
    async def find(self, user_id: str) -> dict:
        return {"id": user_id}

---

streaming_response

def streaming_response(
    content: AsyncGenerator[bytes, None] | Iterator[bytes],
    status_code: int = 200,
    headers: dict[str, str] | None = None,
    media_type: str | None = None
) -> StreamingResponse

Create a streaming response

---

throttle

def throttle(
    rate: str,
    *,
    by: Literal['user', 'ip', 'endpoint'] = 'user'
) -> Callable[[Any], Any]

Rate-limit a route handler using a human-readable rate string.

This is ergonomic sugar over the lower-level rate_limit decorator.

Parameters

Parameter	Type	Description
`rate`	str	Rate string in ``"/"`` format, e.g. ``"30/minute"``, ``"5/hour"``, ``"100/second"``.
`by`	Literal['user', 'ip', 'endpoint']	Scope for the rate limit key: * ``"user"`` — per authenticated user (falls back to IP when ``request.state.user`` is not set). * ``"ip"`` — per client IP address. * ``"endpoint"`` — per route path + HTTP method.

Returns

Type	Description
Callable[[Any], Any]	Decorator that enforces the rate limit on the decorated handler.

Raises

Exception	Description
ValueError	If *rate* is not a valid rate string.

Example

@get("/search")
@throttle("30/minute")
async def search(self, request: Request) -> list[dict]: ...

@post("/upload")
@throttle("5/hour", by="user")
async def upload(self, request: Request) -> dict: ...

---

trace

def trace(
    path: str,
    **kwargs: Any
) -> Any

Register a TRACE route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the route. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured route handler function.

---

version

def version(api_version: str) -> Callable[[Any], Any]

Decorator to specify controller or method version.

Usage

@version("1")
class UsersController(Controller):
    ...

@version("2")
class UsersV2Controller(Controller):
    ...

---

websocket

def websocket(
    path: str,
    **kwargs: Any
) -> Any

Register a WebSocket route handler.

Parameters

Parameter	Type	Description
`path`	str	URL path pattern for the WebSocket endpoint. **kwargs: Additional route configuration.

Returns

Type	Description
Any	Configured WebSocket handler function.

---

websocket_handler

def websocket_handler(
    path: str,
    *,
    ping_interval: int | None = None,
    ping_timeout: int | None = None,
    max_connections_per_user: int | None = None
) -> Any

Decorator to mark a class as a WebSocket handler.

This decorator registers the handler class with the router and configures the WebSocket path.

Parameters

Parameter	Type	Description
`path`	str	URL path for the WebSocket endpoint (can include path parameters)
`ping_interval`	int | None	Override default ping interval
`ping_timeout`	int | None	Override default ping timeout
`max_connections_per_user`	int | None	Override max connections per user

Example

@websocket_handler("/ws/chat/{room}")
class ChatHandler(AbstractWebSocketHandler):
    async def on_connect(self, websocket):
        await websocket.accept()
        await self.broadcast({"type": "join", "user": "..."})

    async def on_message(self, websocket, message):
        await self.broadcast(message)

    async def on_disconnect(self, websocket):
        await self.broadcast({"type": "leave", "user": "..."})

---

Exceptions

ConflictError

409 Conflict.

def __init__(
    detail: str = 'Conflict',
    cause: Exception | None = None
) -> None

---

HTTPError

Base HTTP error, compatible with LexigramError.

def __init__(
    status_code: int,
    detail: str = '',
    headers: dict[str, str] | None = None,
    code: str | None = None,
    cause: Exception | None = None
) -> None

---

InternalServerError

500 Internal Server Error.

def __init__(
    detail: str = 'Internal Server Error',
    code: str = 'INTERNAL_SERVER_ERROR',
    cause: Exception | None = None
) -> None

---

NotFoundError

404 Not Found.

def __init__(
    detail: str = 'Not Found',
    cause: Exception | None = None
) -> None

---

RateLimitError

429 Too Many Requests.

def __init__(
    detail: str = 'Too Many Requests',
    retry_after: int | None = None,
    cause: Exception | None = None
) -> None

---

TooManyConnectionsError

503 Service Unavailable — connection limit reached.

Raised when a streaming endpoint (e.g. SSE) has reached its max_connections cap and cannot accept further connections.

def __init__(
    detail: str = 'Too many active connections',
    cause: Exception | None = None
) -> None

---