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.

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

from quadkit.serialization.domain import json_serialize
from quadkit.web import JSONResponse

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

from quadkit.serialization.domain import json_serialize
from quadkit.web import JSONResponse

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

__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

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

source

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 Quadkit middleware stack as an explicit, composable list.

Encapsulates the minimum set of middlewares that every Quadkit 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 Quadkit middlewares to include in the stack. They are adapted to Starlette ``Middleware` `wrappers via MiddlewareAdapterRegistry and follow` `DIScopeMiddleware`` in user-supplied order — the DI-scope layer always stays outermost so request-scoped resolution is available to every later layer.

Example

from quadkit.web.middleware import DefaultMiddlewareStack
from quadkit.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 quadkit.web.middleware import DefaultMiddlewareStack
from quadkit.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 Quadkit 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 quadkit 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 Quadkit 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 to control the HTTP status code a domain error maps to; the decorator registers the mapping, using the decorated controller as the decoration target

from quadkit.web import error_status

@error_status(ItemNotFound, 404)
class ItemsController(Controller): ...

from quadkit.web import error_status

@error_status(ItemNotFound, 404)
class ItemsController(Controller): ...

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

`GracefulShutdownManager`

source

Manages graceful shutdown with connection draining.

__init__

def __init__(timeout: float = 30.0)

source

is_shutting_down

property is_shutting_down() -> bool

source

Check if shutdown is in progress.

active_connections

property active_connections() -> int

source

Get the number of active connections.

wait_for_drain

async def wait_for_drain() -> None

source

Wait for active connections to complete or timeout.

begin_shutdown

def begin_shutdown() -> None

source

Begin the shutdown process.

complete_shutdown

def complete_shutdown() -> None

source

Complete the shutdown process.

track_connection

async def track_connection() -> Any

source

Context manager to track active connections.

serve_503_during_drain

async def serve_503_during_drain(request: Any) -> Any

source

Return a 503 response during drain period.

`GracefulShutdownProvider`

source

DI provider that coordinates HTTP request draining on shutdown.

__init__

def __init__(
    timeout: float = 30.0,
    shutdown_manager: GracefulShutdownManager | None = None
) -> None

source

manager

property manager() -> GracefulShutdownManager

source

Return the shared graceful shutdown manager.

middleware

property middleware() -> tuple[type[ShutdownMiddleware], dict[str, Any]]

source

Return the ASGI middleware tuple backed by the shared manager.

async def register(container: ContainerRegistrarProtocol) -> None

source

boot

async def boot(container: ContainerResolverProtocol) -> None

source

No-op: shutdown wiring is shared through DI and middleware config.

on_before_shutdown

async def on_before_shutdown(signal: str | None = None) -> None

source

Begin HTTP drain and wait for in-flight requests to complete.

shutdown

async def shutdown() -> None

source

No-op: request drain already completed in on_before_shutdown().

`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

`HTMXResponse`

source

Convenience HTML response with HTMX-specific headers support.

Example

return HTMXResponse(“

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

`HealthController`

source

HTTP probe controller exposing ``/health``, ``/ready``, ``/live``, ``/startup``, and ``/metrics``.

__init__

def __init__(
    registry: HealthCheckRegistryProtocol,
    metrics_exporter: MetricsExpositionProtocol | None = None
) -> None

source

health_check

async def health_check() -> JSONResponse

source

Basic liveness summary probe.

readiness

async def readiness() -> JSONResponse

source

Readiness probe.

liveness

async def liveness() -> JSONResponse

source

Liveness probe.

startup

async def startup() -> JSONResponse

source

Startup probe.

metrics

async def metrics() -> Response

source

Export Prometheus-formatted metrics when a collector/exporter is wired.

`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: ASGIApp,
    sanitize_query_params: bool = True
) -> None

source

`JSONResponse`

source

JSON response with high performance.

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

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

json_schema

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

source

Default wire shape from to_dict; extension members remain allowed.

from_exception

def from_exception(
    cls,
    exc: Exception,
    status: int = 500,
    debug: bool = False,
    **kwargs: Any
) -> 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: RedisClientProtocol | None = None,
    enabled: bool = True
) -> 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 validated at module configuration time. Scope note: global (all-routes) enforcement lives in the ASGI ``RateLimitMiddleware, configured through the` `[web.rate_limit]` `section of` `application.yaml`` — this kwarg does not wire that middleware; per-route limits are enforced by throttle with the module-provided limiter.
`redis_client`	RedisClientProtocol \| None	Pre-built Redis client satisfying RedisClientProtocol. Only used when ``backend="redis"``. When None, the provider resolves one from the container at boot time.
`enabled`	bool	When ``False`` the rate limiter is not started.

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.

A client reaches the container as a RedisClientProtocol registration or as a named "redis_client" binding; see resolve_redis_client.

__init__

def __init__(
    redis_client: RedisClientProtocol | None = None,
    enabled: bool = True,
    backend: Literal['redis', 'memory'] = 'redis'
) -> None

source

Initialize rate limit provider.

Parameters

Parameter	Type	Description
`redis_client`	RedisClientProtocol \| None	Optional pre-built client satisfying RedisClientProtocol.
`enabled`	bool	When ``False`` the provider skips all registration/boot.
`backend`	Literal['redis', 'memory']	``"redis"` `uses the Redis-backed limiter (previous and default behaviour);` `"memory"`` binds the process-local sliding-window MemoryLimiter and never touches Redis.

async def register(container: ContainerRegistrarProtocol) -> None

source

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

`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 `DomainError`	400 (client fault)
non-domain error	500 (server fault)

Use register to override the status and optionally attach an app-specific body renderer for a specific error type.

to_response

def to_response(
    result: Any,
    success_status: int = 200,
    *,
    request: Any | None = None
) -> StarletteResponse

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).
`request`	Any \| None	Optional request context passed to a registered error renderer.

Returns

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

documented_errors

def documented_errors(
    cls,
    error_types: list[type[Exception]] | tuple[type[Exception], Ellipsis],
    *,
    exclude_statuses: set[str] | None = None
) -> dict[str, Any]

source

Describe explicitly declared default Result errors using native wiring.

Custom renderers require explicit responses; no return-annotation or guard-name guessing, exception construction or policy inference occurs.

def register(
    cls,
    error_type: type[Exception],
    status_code: int,
    *,
    body_renderer: ErrorBodyRenderer | None = None
) -> None

source

Custom status registrations take precedence over defaults (inserted at the front). A body renderer receives (error, status_code, request) and may return either a JSON-serializable body or a Starlette Response. When multiple renderers match an error, the nearest type in its MRO wins; an unmatched error keeps the default RFC 7807 body.

Parameters

Parameter	Type	Description
`error_type`	type[Exception]	The exception class to match.
`status_code`	int	HTTP status code to return when the error does not provide its own ``status_code`` attribute.
`body_renderer`	ErrorBodyRenderer \| None	Optional app-specific error body renderer. Returning a Response allows an app to preserve response headers or media types alongside its body; returning data lets the mapper apply the resolved status as JSON.

error_to_response

def error_to_response(
    cls,
    error: Any,
    request: Any | None = None
) -> StarletteResponse

source

Convert a domain error to a JSON HTTP response.

Parameters

Parameter	Type	Description
`error`	Any	The error value from ``Result.unwrap_err()``.
`request`	Any \| None	Optional request context passed to a registered renderer.

Returns

Type	Description
StarletteResponse	The response returned by a custom renderer, or a JSONResponse with the appropriate status code and a structured RFC 7807 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

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,
    global_pipeline: FilterPipeline | None = None
)

source

Initialize the router.

Parameters

Parameter	Type	Description
`filter_pipeline`	FilterPipeline \| None	Pipeline holding this router's own filters.
`global_pipeline`	FilterPipeline \| None	Application-wide pipeline owned by the DI container; passed down to each request pipeline as the last filter consulted before an exception is re-raised.

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)

`ShutdownMiddleware`

source

Middleware that tracks connections for graceful shutdown.

__init__

def __init__(
    app: ASGIApp,
    shutdown_manager: GracefulShutdownManager
) -> None

source

`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 Quadkit 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 api_docs: API documentation endpoints and their assets enable_identity_resolution: Resolve OAuth external IDs to internal UUIDs enable_auth: Enable built-in authentication middleware auth_exclude_paths: Paths excluded from authentication

api_docs_enabled

property api_docs_enabled() -> bool

source

Whether the documentation endpoints are served in this environment.

api_docs.enabled is tri-state: None (the default) serves the documentation everywhere except production, True opts in unconditionally — including production — and False withholds it everywhere. Both the endpoints (/docs, /redoc, /openapi.json) and their assets follow this one answer.

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,
    websocket_handlers: list[type] | 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 decorated WebSocket handlers, merging them with explicit 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 quadkit.web import WebProvider, WebConfig

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

from quadkit.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],
    filters=[AuditExceptionFilter()],
    web_config=WebConfig(debug=True),
))

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

filters are application-wide exception filters. They are registered as the container’s FilterPipeline singleton and consulted after the route-level filters and the router’s built-in defaults, so a filter listed here is the last stop before an error propagates.

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,
    filters: 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,
    websocket_handlers: list[type] | None = None
) -> None

source

contributor_registry

property contributor_registry() -> Any

source

Expose contributor registry for route setup access.

route_registry

property route_registry() -> Any

source

This application’s route registry (per-instance, plans/09172026 H-04).

controller_registry

property controller_registry() -> Any

source

This application’s controller registry (per-instance, H-04).

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 routes auto-discovered from packages.

Scans each package recursively for Controller subclasses and @websocket_handler classes, registering both 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"))

async def register(container: ContainerRegistrarProtocol) -> None

source

Delegates each cohesive registration group to quadkit.web.di.provider_sections, invoked at the same ordered positions as the pre-extraction implementation.

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 synchronously (blocking).

Delegates to quadkit.web.server.runner.run_server, which is sync-only; call quadkit.web.server.runner.run_server_async from async code instead.

`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

Quadkit WebSocket wrapper around Starlette WebSocket

__init__

def __init__(starlette_ws: StarletteWebSocket)

source

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

source

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

source

receive_text

async def receive_text() -> str

source

receive_json

async def receive_json() -> Any

source

receive_event

async def receive_event() -> MutableMapping[str, Any]

source

Receive a raw ASGI WebSocket event for protocol bridges.

Starlette types the ASGI message as a mutable mapping, not a plain dict; declaring the wider type is what keeps make type honest without an unchecked cast.

headers

property headers() -> Any

source

Handshake headers exposed read-only to gateway handlers.

query_params

property query_params() -> Any

source

Handshake query parameters exposed to gateway handlers.

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.

path_params

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

source

Path parameters captured by the Starlette WebSocket route.

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 register a controller class with the app being configured.

Registers the class by name in the active application’s ControllerRegistry (plans/09172026 H-04: registries are per-application instances — there is no module-level global to write to). When no registration context is active the class is returned unchanged and a warning is logged; declare controllers via WebProvider(controllers=...) or @module(controllers=...) to have them mounted.

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

source

Cookie parameter decorator.

`delete`

delete

def delete(
    path: str,
    **kwargs: Any
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	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).

`discover_websocket_handlers`

discover_websocket_handlers

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

source

Scan packages for classes marked with ``@websocket_handler``.

WebSocket handlers are deliberately discovered separately from HTTP controllers: they have a class-level path marker and are mounted as Starlette WebSocketRoute instances rather than HTTP routes.

Parameters

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

Returns

Type	Description
list[type]	Deduplicated handler classes in discovery order.

`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
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	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
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	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 quadkit.web import app, get, injectable
from quadkit.logging import get_logger

logger = get_logger(__name__)

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

from quadkit.web import app, get, injectable
from quadkit.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
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	Configured route handler function.

`patch`

patch

def patch(
    path: str,
    **kwargs: Any
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	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
) -> Callable[[F], F]

source

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
Callable[[F], F]	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
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	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

Registers the controller in both the active RouteRegistry and ControllerRegistry, ensuring synchronization and preventing lookup inconsistencies. With no active registration_context this is a no-op (with a warning) — import-time side effects on shared module state are gone; declare controllers via WebProvider(controllers=...) or @module(controllers=...) instead.

`render_template`

render_template

def render_template(
    name: str,
    context: dict[str, Any] | None = None,
    templates: Jinja2Templates | None = None,
    **kwargs: Any
) -> 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.

Note

Guarded handlers must declare request: Request as a parameter — the guard locates the principal on request.state.user at runtime (stamp it from your authentication middleware; see AuthenticatorProtocol).

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 quadkit.di.singleton DI marker (which sets __quadkit_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 quadkit.web import app, get, singleton

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

from quadkit.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
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	Configured route handler function.

`use_interceptors`

use_interceptors

def use_interceptors(*interceptors: Any) -> Any

source

Attach one or more interceptors to a route handler or controller.

Interceptors wrap handler execution: code before await next_handler.handle() runs on the way in; code after runs on the way out and may transform the result. Applied to a controller class, the interceptors cover every route the class declares; handler-level interceptors take precedence.

Parameters

Parameter	Type	Description

Example

@use_interceptors(EnvelopeInterceptor())
class ReportsController(Controller): ...

@use_interceptors(AuditLogInterceptor())
async def sensitive_action(self): ...

@use_interceptors(EnvelopeInterceptor())
class ReportsController(Controller): ...

@use_interceptors(AuditLogInterceptor())
async def sensitive_action(self): ...

`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
) -> Callable[[F], F]

source

Parameters

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

Returns

Type	Description
Callable[[F], F]	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 QuadkitError.

__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