API Reference

Protocols

CallHandlerProtocol

source

Wraps the next step in the pipeline.

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

handle

async def handle() -> Any

source

Continue processing the request.

Returns the result of the next handler in the pipeline.

---

ConfigAccessorProtocol

source

Provides access to web configuration.

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

web_config

property web_config() -> Any

source

Return the web-layer configuration.

Returns

Type	Description
Any	WebConfig instance with web-layer settings.

provider_config

property provider_config() -> Any

source

Return the provider-specific configuration.

Returns

Type	Description
Any	WebProviderConfig instance with provider settings.

debug_routes_auth

property debug_routes_auth() -> Any

source

Return the debug routes authentication handler, if set.

Returns

Type	Description
Any	Callable or None.

fail_on_route_conflict

property fail_on_route_conflict() -> bool

source

Whether to raise on duplicate route registration.

Returns

Type	Description
bool	True if duplicate routes should raise RuntimeError.

---

ControllerSourceProtocol

source

Provides the list of registered controller classes.

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

controllers

property controllers() -> list[type]

source

Return the list of registered controller classes.

Returns

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

---

ExecutionContextProtocol

source

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.

request

property request() -> Any

source

The current HTTP request.

handler

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

source

The handler function that will be called.

controller_class

property controller_class() -> type | None

source

The controller class, if this is a controller method.

method_name

property method_name() -> str | None

source

The name of the method being called.

route_metadata

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

source

Metadata about the route.

---

PipeProtocol

source

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}")

transform

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

source

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

source

Provides access to internal provider resources.

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

router

property router() -> Any

source

Return the internal Router instance.

Returns

Type	Description
Any	Router instance or None if not initialized.

openapi_generator

property openapi_generator() -> Any

source

Return the OpenAPI generator instance.

Returns

Type	Description
Any	OpenAPIGenerator instance or None if not configured.

---

WebAppAccessorProtocol

source

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.

starlette

property starlette() -> Any

source

Return the Starlette application instance.

Returns

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

---

WebInterceptorProtocol

source

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

intercept

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

source

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

source

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

source

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

url_prefix

property url_prefix() -> str

source

Return the URL prefix for this version.

---

BackgroundTasks

source

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.

__init__

def __init__() -> None

source

add

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

source

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

source

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())

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())

__init__

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

source

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.

dispatch_command

async def dispatch_command(command: Any) -> Any

source

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.

dispatch_query

async def dispatch_query(query: Any) -> Any

source

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

source

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 []

__init__

def __init__() -> None

source

collect_routes

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

source

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

source

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).

__init__

def __init__() -> None

source

register

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

source

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

get

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

source

Get a controller class by name.

list_controllers

def list_controllers() -> list[str]

source

List all registered controller names.

get_all_controllers

def get_all_controllers() -> list[type]

source

Get all registered controller classes.

clear

def clear() -> None

source

Clear all registrations.

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

---

DefaultMiddlewareStack

source

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()],
)

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()],
)

__init__

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

source

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.

build

def build() -> list[StarletteMiddleware]

source

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

source

__init__

def __init__(error: E) -> None

source

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

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

source

map_sync

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

source

map_err

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

source

and_then_sync

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

source

or_else_sync

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

source

expect

def expect(message: str) -> T

source

match

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

source

map

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

source

async_map

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

source

Alias for map — exists for backward compatibility.

and_then

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

source

or_else

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

source

flatten

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

source

filter

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

source

ok_or

def ok_or(default: U) -> U

source

---

FastJSONResponse

source

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).

render

def render(content: Any) -> bytes

source

---

FileResponse

source

File response for serving static files

---

GenericController

source

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): ...

from lexigram.web import error_status

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

__init__

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

source

resource_name

property resource_name() -> str

source

Get the resource name for permissions and error messages.

logger

property logger() -> Any

source

Get the logger for this controller.

list_items

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

source

List items with pagination.

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

get_item

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

source

Get single item by ID.

create_item

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

source

Create new item (returns 201 on success).

update_item

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

source

Update existing item.

delete_item

async def delete_item(item_id: str) -> Any

source

Delete item (returns 204 on success).

---

HTMLContent

source

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

source

HTML response

__init__

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

source

---

HTMXResponse

source

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.

__init__

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

source

---

InputSanitizationMiddleware

source

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``).

__init__

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

source

---

JSONResponse

source

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.

__init__

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

source

---

Ok

source

__init__

def __init__(value: T) -> None

source

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

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

source

map_sync

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

source

map_err

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

source

and_then_sync

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

source

or_else_sync

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

source

expect

def expect(message: str) -> T

source

match

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

source

map

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

source

async_map

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

source

Alias for map — exists for backward compatibility.

and_then

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

source

or_else

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

source

flatten

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

source

filter

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

source

ok_or

def ok_or(default: U) -> T

source

---

ParamMetadata

source

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

source

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)

to_dict

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

source

Convert to dictionary for JSON serialization.

from_exception

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

source

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.

validation_error

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

source

Create a validation error ProblemDetail.

not_found

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

source

Create a not found ProblemDetail.

bad_request

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

source

Create a bad request ProblemDetail.

internal_error

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

source

Create an internal error ProblemDetail.

---

RateLimitModule

source

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"))

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

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

configure

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

source

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

source

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.

__init__

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

source

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.

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register rate limiter in the DI container.

Parameters

Parameter	Type	Description
`container`	ContainerRegistrarProtocol	DI container registrar.

boot

async def boot(container: ContainerResolverProtocol) -> None

source

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

shutdown

async def shutdown() -> None

source

Shutdown the rate limit provider.

health_check

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

source

Check rate-limit provider health.

Returns

Type	Description
HealthCheckResult	HealthCheckResult reflecting whether the limiter is initialised.

---

RedirectResponse

source

Redirect response

---

Request

source

Lexigram request wrapper around Starlette request

__init__

def __init__(starlette_request: StarletteRequest)

source

method

property method() -> str

source

url

property url() -> Any

source

headers

property headers() -> Any

source

query_params

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

source

cookies

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

source

client

property client() -> Any

source

Client address information

app

property app() -> Any

source

Forward to the underlying Starlette app

session

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

source

Access the session state

path_params

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

source

json

async def json() -> Any

source

body

async def body() -> bytes

source

form

async def form() -> Any

source

state

property state() -> Any

source

Request state for middleware

scope

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

source

Access the underlying ASGI scope

request

property request() -> StarletteRequest

source

Access the underlying Starlette request

user

property user() -> Any | None

source

Authenticated user set by auth middleware.

auth

property auth() -> Any | None

source

Authentication credentials.

request_id

property request_id() -> str | None

source

Request ID set by RequestIDMiddleware.

is_json

property is_json() -> bool

source

Check if request content type is JSON.

accepted_types

property accepted_types() -> list[str]

source

Parse Accept header into ordered list.

ip

property ip() -> str

source

Client IP address (proxy-aware).

validated_data

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

source

Data validated/transformed by pipe pipeline.

validated_data

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

source

Set validated data from pipe pipeline.

---

Response

source

Base response class

---

Result

source

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

is_ok

def is_ok() -> bool

source

is_err

def is_err() -> bool

source

unwrap

def unwrap() -> T

source

unwrap_err

def unwrap_err() -> E

source

unwrap_or

def unwrap_or(default: T) -> T

source

unwrap_or_else

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

source

map_sync

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

source

map_err

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

source

and_then_sync

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

source

or_else_sync

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

source

expect

def expect(message: str) -> T

source

match

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

source

map

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

source

and_then

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

source

or_else

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

source

flatten

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

source

filter

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

source

ok_or

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

source

from_exception

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

source

Wrap a caught exception into an Err result.

to_optional

def to_optional() -> T | None

source

inspect

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

source

inspect_err

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

source

---

ResultResponseMapper

source

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

to_response

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

source

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.

register

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

source

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.

error_to_response

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

source

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

source

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

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

Usage

__init__

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

source

can_activate

async def can_activate(context: Any) -> bool

source

Check if user has required role using AuthorizerProtocol.

---

RouteRegistry

source

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).

__init__

def __init__(debug: bool = False)

source

register_controller

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

source

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.

get_all_routes

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

source

Get all registered routes

get_controllers

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

source

Get all registered controllers

find_route

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

source

Find a route by path and method

get_routes_by_controller

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

source

Get all routes for a specific controller

clear

def clear() -> None

source

Clear all registrations

---

Router

source

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.

__init__

def __init__(filter_pipeline: FilterPipeline | None = None)

source

get_route_by_name

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

source

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.

clear_cache

def clear_cache() -> None

source

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)

hot_reload_manager.on_reload(router.clear_cache)

add_route

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

source

Add a route (used internally)

---

StreamingResponse

source

Streaming response for large files or server-sent events

---

VersionExtractor

source

Extracts version from requests based on strategy.

__init__

def __init__(config: VersioningConfig)

source

extract

def extract(request: Request) -> str

source

Extract version from request.

---

WebConfig

source

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

validate_production_security

def validate_production_security() -> WebConfig

source

Block insecure configurations in production.

---

WebInterceptorBase

source

Base class for interceptors (optional convenience).

Provides a no-op implementation that subclasses can override.

intercept

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

source

Default implementation just passes through.

---

WebModule

source

Web module with HTTP provider.

configure

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

source

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.

stub

def stub(
    cls,
    config: Any = None
) -> DynamicModule

source

Return a no-op WebModule suitable for unit testing.

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

Parameters

Parameter	Type	Description
`config`	Any	Optional test configuration override.

Returns

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

---

WebProvider

source

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())

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, ...)))

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),
))

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.

__init__

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

source

contributor_registry

property contributor_registry() -> Any

source

Expose contributor registry for route setup access.

from_config

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

source

Create a WebProvider from config.

Context kwargs may include middleware, controllers, exception_handlers.

auto_discover

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

source

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"))

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

register

async def register(container: ContainerRegistrarProtocol) -> None

source

Register web services in DI container

boot

async def boot(container: ContainerResolverProtocol) -> None

source

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.

shutdown

async def shutdown() -> None

source

Cleanup resources created by this provider.

health_check

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

source

Check web provider health.

run_server

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

source

Run the web application using Granian.

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

---

WebRequestReceivedHook

source

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

source

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

source

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

---

WebServerStoppedHook

source

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

---

WebSocket

source

Lexigram WebSocket wrapper around Starlette WebSocket

__init__

def __init__(starlette_ws: StarletteWebSocket)

source

accept

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

source

close

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

source

receive_text

async def receive_text() -> str

source

receive_json

async def receive_json() -> Any

source

send_text

async def send_text(data: str) -> None

source

send_json

async def send_json(data: Any) -> None

source

send_bytes

async def send_bytes(data: bytes) -> None

source

state

property state() -> Any

source

WebSocket state for middleware

---

Functions

api_version

api_version

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

source

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

@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

background

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

source

Decorator to run a function as a background task.

---

body

body

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

source

Request body parameter decorator.

---

controller

controller

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

source

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

cookie

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

source

Cookie parameter decorator.

---

delete

delete

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

source

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

discover_controllers

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

source

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

error_status

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

source

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):
    ...

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

---

file

file

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

source

File upload parameter decorator.

---

file_response

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

source

Create a file response

---

form

form

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

source

Form parameter decorator.

---

get

get

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

source

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

get_current_user_optional

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

source

Get current user from state (optional).

---

get_current_user_required

get_current_user_required

async def get_current_user_required(request: Request) -> Any

source

Get current user from state (required).

---

get_request_id

get_request_id

async def get_request_id(request: Request) -> str

source

Get request ID from state.

---

get_version

get_version

def get_version(request: Request) -> str

source

Get API version from request state.

Parameters

Parameter	Type	Description
`request`	Request	The HTTP request

Returns

Type	Description
str	API version string

---

guard

guard

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

source

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): ...

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

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

---

head

head

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

source

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

header

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

source

Header parameter decorator.

---

html_response

html_response

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

source

Create an HTML response

---

injectable

injectable

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

source

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)

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

json_response

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

source

Create a JSON response

---

options

options

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

source

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

patch

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

source

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

path

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

source

Path parameter decorator.

---

post

post

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

source

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

put

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

source

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

query

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

source

Query parameter decorator.

---

redirect_response

redirect_response

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

source

Create a redirect response

---

register_controller

register_controller

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

source

Convenience function to register a controller.

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

---

render_template

render_template

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

source

---

roles

roles

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

source

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): ...

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

singleton

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

source

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}

from lexigram.web import app, get, singleton

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

---

streaming_response

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

source

Create a streaming response

---

throttle

throttle

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

source

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: ...

@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

trace

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

source

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

version

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

source

Decorator to specify controller or method version.

Usage

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

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

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

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

---

websocket

websocket

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

source

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

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

source

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

source

409 Conflict.

__init__

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

source

---

HTTPError

source

Base HTTP error, compatible with LexigramError.

__init__

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

source

---

InternalServerError

source

500 Internal Server Error.

__init__

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

source

---

NotFoundError

source

404 Not Found.

__init__

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

source

---

RateLimitError

source

429 Too Many Requests.

__init__

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

source

---

TooManyConnectionsError

source

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.

__init__

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

source

---