without_web¶
An opinionated HTTP/WebSocket router for without-asgi: trie matching, typed path params, 405-vs-404, mounting, and OpenAPI.
module-attribute
¶
module-attribute
¶
module-attribute
¶
module-attribute
¶
module-attribute
¶
UUID: Converter[UUID] = Converter(
name="uuid",
parse=uuid.UUID,
schema={"type": "string", "format": "uuid"},
)
WebsocketExceptionRecover = Callable[
[Exception], Awaitable[WebsocketClose | None]
]
dataclass
¶
Bases: Generic[_V_co]
A path-segment parser paired with the JSON Schema it parses into.
parse turns a single matched segment into a typed value, raising
ValueError to reject a segment that does not fit (int against "abc").
Rejection is not a handler-side error: it makes that trie branch fail to
match so the walk backtracks to a sibling, ultimately a 404 if nothing
matches (parse, don't validate).
schema is the half the router contributes to OpenAPI for a path parameter
that uses this converter: the router owns the path-param schema because it
owns the converter.
name is the converter's identity (and its OpenAPI parameter style); a
typed-token pattern reuses the converter value directly
(path_param("id", INT)), so the name, parse, type, and schema are all
declared in one place. Equality and hashing are by name alone (a converter
is a trie key), so parse and schema are excluded from comparison.
dataclass
¶
BufferedRequest(
scope: HttpScope,
path_params: Mapping[str, object],
query_params: Mapping[str, tuple[str, ...]],
body: bytes,
)
Bases: HttpRequestHead
An HttpRequestHead plus the fully-buffered request body, built only on the
buffered-HTTP path. The body extractor's context is exactly this type, so a
body token on a streaming or websocket route (which build a bodyless
HttpRequestHead/WebsocketRequestHead) is a static type error, not a runtime
guard.
Bases: ValueError
A request rejected while one of its typed values was being extracted.
The reject signal an extractor raises when a parse (a once/optional
cardinality check, a converter, a pydantic model) refuses the input. It gathers
at the raise site everything a recover policy needs to answer well: field
names the request part that failed (a query/header parameter name, or None for
the body), and cause carries the underlying error as a first-class value, so a
policy matches case ExtractionError(cause=ValidationError()) to answer a 422
for an invalid body versus a 400 for a bad query/header value, without reaching
into __cause__.
A ValueError is the codebase's "reject" signal (the same one a converter
raises to backtrack the trie walk), so the extractors turn one into a rich
ExtractionError, and the router wraps any stray one left unattributed. Making
the boundary a single matchable type is what lets a plain ValueError raised
deeper in a handler still surface as a 500 rather than masquerading as a client
error.
dataclass
¶
Extractor(
extract: Callable[[_C_contra], _V_co],
query: tuple[QueryParam, ...] = (),
headers: tuple[HeaderParam, ...] = (),
request_body: Body | None = None,
path: PathSpec | None = None,
)
Bases: Generic[_C_contra, _V_co]
A typed piece of a request, paired with the OpenAPI it contributes.
extract is a pure C -> V (where C is the request context it reads: the
permissive RequestHead, or a narrower HttpRequestHead/WebsocketRequestHead/
BufferedRequest) that raises to reject a bad request (a catching
middleware's recover maps the raised type to a 4xx); it never decides which
handler runs. The context parameter is what lets a handler refuse the wrong
extractor for its route kind statically (a body on a streaming route, an
http_scope on a websocket). The same value carries its own OpenAPI fragment,
so a handler's parameter list and request body are recovered from the
extractors it declares, never restated: one declaration, two consumers
(parse and describe).
dataclass
¶
HttpRequestHead(
scope: HttpScope,
path_params: Mapping[str, object],
query_params: Mapping[str, tuple[str, ...]],
)
Bases: RequestHead
A RequestHead whose scope is known to be an HttpScope: the context of any
HTTP route (buffered or streaming). http_scope() reads its narrowed scope
with no runtime check, and the streaming-HTTP path builds it directly.
dataclass
¶
RequestHead(
scope: HttpScope | WebsocketScope,
path_params: Mapping[str, object],
query_params: Mapping[str, tuple[str, ...]],
)
The parsed head of a request (or websocket handshake): everything an extractor
reads except the body. The read-only, parsed-once context handed to each
scope-derived extractor, and the most general context in the lattice below:
path_param, catch_all, query_param, and header_param read only what is
here, so they work on any route.
path_params holds the path parameters the router already parsed during the
trie walk (typed values, stored as object); query_params is the query
string decoded and parsed once (via parse_qs), each name's values held as an
immutable tuple, so a handler declaring N query_param tokens shares one parse
rather than re-decoding the query string per token. scope is
HttpScope | WebsocketScope so a query_param/header_param token reads either
(both carry query_string and headers); the whole-scope read is split into the
protocol-specific http_scope()/websocket_scope(), since only those know the
concrete type. A header_param reads the scope's raw header pairs directly
through the without_asgi.headers functions.
The subtypes narrow the two facts a route fixes, so the wrong extractor on the
wrong route is a static error rather than a runtime guard (see Extractor):
HttpRequestHeadnarrowsscopetoHttpScope(whathttp_scope()needs).WebsocketRequestHeadnarrowsscopetoWebsocketScope(websocket_scope()).BufferedRequest(anHttpRequestHead) adds the bufferedbody(body()).
Each route builds exactly its concrete context: the buffered-HTTP path a
BufferedRequest, the streaming-HTTP path an HttpRequestHead, a websocket a
WebsocketRequestHead. This base is never built directly; it names the shared
top so the permissive extractors can slot into all three.
dataclass
¶
WebsocketRequestHead(
scope: WebsocketScope,
path_params: Mapping[str, object],
query_params: Mapping[str, tuple[str, ...]],
)
Bases: RequestHead
A RequestHead whose scope is known to be a WebsocketScope: the context of a
websocket route. websocket_scope() reads its narrowed scope with no runtime
check, and the websocket path builds it directly.
classmethod
¶
parsed(
scope: WebsocketScope, path_params: Mapping[str, object]
) -> WebsocketRequestHead
Assemble a WebsocketRequestHead, parsing the scope's query string once at the boundary.
dataclass
¶
dataclass
¶
dataclass
¶
dataclass
¶
RouteSpec(
summary: str = "",
query: tuple[QueryParam, ...] = (),
headers: tuple[HeaderParam, ...] = (),
request_body: Body | None = None,
responses: Mapping[int, ResponseSpec] = dict(),
)
The handler-owned half of an endpoint's OpenAPI description.
The router never sees the body or interprets the query, so it cannot be the
source of those schemas: an endpoint declares them here, in the one place
they are also parsed. openapi merges this with the router's
path/method/path-param half.
dataclass
¶
Sequence(item_schema: SchemaRef)
Body content that is a sequence of items, each validating against item_schema.
Renders as OpenAPI 3.2's itemSchema, the description of a sequential media
type (NDJSON, JSON Lines, application/json-seq, SSE text/event-stream,
multipart/mixed, ...). without-web is agnostic to the framing on the wire:
the application names it via Body.media_type and emits the bytes itself.
This is documentation only; nothing on the runtime path reads it.
dataclass
¶
dataclass
¶
dataclass
¶
How a path-param extractor appears as a route segment.
The bridge that lets one path_param(...) value be both a pattern segment
(the router matches and schemas it through converter) and a typed read in
the handler. name binds the segment; catch_all marks the rest-consuming
form.
dataclass
¶
Delegate(prefix: str, target: HttpRouter[T])
An opaque HTTP sub-application delegated to at a literal-string prefix.
The bring-your-own-app escape hatch: target (another HttpRouter, a legacy
app) is handed the prefix-trimmed scope (ASGI root_path semantics) and
treated as a black box, since its routes cannot be seen, baked, or reversed.
Transparent sub-apps whose routes you own use mount(...) instead, which
bakes the prefix into the routes so they stay first-class values. The prefix
is a plain str, hence a literal path with no parameter to bind.
dataclass
¶
What the router hands a handler: the scope plus already-parsed path params.
make_asgi_app's HttpRouter type is unchanged: Router.dispatch still
presents as (T, HttpScope) -> HttpHandler. The richer Match is the
router's internal endpoint protocol, the one place a handler reads the
path parameters the route pattern bound.
Bases: Protocol
Anything reverse routing can render: a value carrying parsed path segments.
Both Route and WebsocketRoute satisfy it structurally, so url_for takes
either without naming the lifespan-state type they are generic over (which
would otherwise fight variance).
dataclass
¶
Route(
segments: tuple[Segment, ...],
methods: Mapping[str, HttpEndpoint[T]],
)
A route: parsed path segments bound to one endpoint per HTTP method.
The method-decorator form (@get(...)) produces a single-method Route; the
Router merges Routes that share a path into one method map, so the
405-vs-404 split still falls out of the trie. segments is the complete
path, mount prefixes already baked in (see mount), so a Route is a
self-contained value: it reverses (url_for) with no router, and its meaning
does not depend on where it is placed.
dataclass
¶
Router(
routes: tuple[Route[T] | Delegate[T], ...],
fallback: HttpEndpoint[T],
middleware: HttpMiddleware[T] = _PASSTHROUGH_HTTP,
)
An opinionated HTTP router whose dispatch is an HttpRouter[T].
The whole integration surface with without-asgi is that one type: pass
router.dispatch as make_asgi_app(http=...) and bring-your-own (or no
router at all) stays first-class. The route table is compiled to an
immutable trie once at construction; dispatch is then a pure walk that
recovers route precedence, 405-vs-404, and delegation from the tree's shape.
Routes are flat, self-contained values (mount prefixes are baked in by
mount); only opaque Delegates stay as wrappers, since a black box cannot
be flattened.
dataclass
¶
WebsocketDelegate(prefix: str, target: WebsocketRouter[T])
The WebSocket sibling of Delegate: an opaque WebSocket app at a prefix.
Handed the prefix-trimmed scope and treated as a black box; transparent
WebSocket sub-apps use ws_mount(...) to bake the prefix into their routes.
dataclass
¶
WebsocketRoute(
segments: tuple[Segment, ...],
endpoint: WebsocketEndpoint[T],
)
Parsed path segments bound to one WebSocket endpoint (the sibling of Route).
dataclass
¶
WebsocketRouter(
routes: tuple[
WebsocketRoute[T] | WebsocketDelegate[T], ...
],
fallback: WebsocketEndpoint[T],
middleware: WebsocketMiddleware[
T
] = _PASSTHROUGH_WEBSOCKET,
)
The WebSocket sibling of Router, reusing the same trie machinery.
There is no method layer, so no 405: a connection either matches a path or
falls to the fallback. dispatch is a WebsocketRouter[T] for
make_asgi_app(websocket=...). Routes are flat values (prefixes baked by
ws_mount); only opaque WebsocketDelegates stay as wrappers.
catching(
recover: ExceptionRecover,
) -> HttpMiddleware[object]
Build middleware that maps exceptions to a response, before the status commits.
Exception handling is not a new mechanism: it is a Middleware that wraps a
handler and watches its outbound stream. recover is the app's policy: it is
handed a raised exception and returns the Response to send instead, or
None to let the exception propagate. There is deliberately no registry of
type -> handler: a recover written as match exc: narrows each case to
its real type (no assert isinstance) and can re-raise, chain, or do async
work, all of which a heterogeneous mapping could not express without a cast.
Honest limitation: the mapping applies only while the status line can still
be set, i.e. until the first ResponseStart flows out. Informational events
(EarlyHint, ResponseDebug) precede it and do not commit the status, so an
exception after them can still be mapped; once ResponseStart is on the
wire the exception re-raises, because the handler can abort but not re-status.
catching_websocket(
recover: WebsocketExceptionRecover,
) -> WebsocketMiddleware[object]
The WebSocket sibling of catching, mapping exceptions to a close.
The equivalent commit point is WebsocketAccept: before the handshake is
accepted a close still rejects the connection (the server turns it into a
403), so a mapped exception becomes that WebsocketClose. Once accepted
the connection is established and the exception re-raises. recover returns
the WebsocketClose to send, or None to propagate.
body(
parse: Callable[[bytes], V],
*,
schema: SchemaRef,
media_type: str = "application/json",
) -> Extractor[BufferedRequest, V]
Parse the buffered request body into V.
parse is injected so without-web stays serialization-agnostic: an app
passes a pydantic model's model_validate_json, a dataclass loader, or any
bytes -> V, and the matching schema is this value's OpenAPI request body. A
parse that raises to reject (a pydantic ValidationError, say) becomes an
ExtractionError with no field (the body is unnamed), the original on cause.
The buffered body lives on BufferedRequest, so that is this extractor's
context: a body token on a bodyless streaming or websocket route is a static
type error (its context is not the HttpRequestHead/WebsocketRequestHead those
routes provide), not a runtime guard.
A typed catch-all path parameter: the {name:path} form as a token.
Consumes the rest of the target into one segment (always the final one); the
sibling of path_param for the rest-of-path case.
header_param(
name: str,
parse: Callable[[tuple[bytes, ...]], V],
*,
schema: SchemaRef,
required: bool = False,
) -> Extractor[RequestHead, V]
Parse a request header into V, given all of its raw values.
Header names are matched case-insensitively; parse receives every value sent
under name as an immutable tuple, in order, and returns V or raises a
ValueError to reject, which becomes an ExtractionError naming this name.
http_scope() -> Extractor[HttpRequestHead, HttpScope]
Hand an HTTP handler the unparsed HttpScope.
The escape hatch that keeps "pass the scope down" and "parse parts of it"
from competing: a handler composes http_scope() alongside parsed extractors
and gets the raw connection facts as just another typed argument. Its context is
HttpRequestHead, whose scope is already an HttpScope, so it reads it with
no runtime check; using it on a websocket route (a WebsocketRequestHead) is a
static type error.
into(
make: Callable[[A, B, C], M],
a: Extractor[R, A],
b: Extractor[R, B],
c: Extractor[R, C],
) -> Extractor[R, M]
into(
make: Callable[[A, B, C, D], M],
a: Extractor[R, A],
b: Extractor[R, B],
c: Extractor[R, C],
d: Extractor[R, D],
) -> Extractor[R, M]
into(
make: Callable[[A, B, C, D, E], M],
a: Extractor[R, A],
b: Extractor[R, B],
c: Extractor[R, C],
d: Extractor[R, D],
e: Extractor[R, E],
) -> Extractor[R, M]
into(
make: Callable[[A, B, C, D, E, F], M],
a: Extractor[R, A],
b: Extractor[R, B],
c: Extractor[R, C],
d: Extractor[R, D],
e: Extractor[R, E],
f: Extractor[R, F],
) -> Extractor[R, M]
into(
make: Callable[[A, B, C, D, E, F, G], M],
a: Extractor[R, A],
b: Extractor[R, B],
c: Extractor[R, C],
d: Extractor[R, D],
e: Extractor[R, E],
f: Extractor[R, F],
g: Extractor[R, G],
) -> Extractor[R, M]
into(
make: Callable[[A, B, C, D, E, F, G, H], M],
a: Extractor[R, A],
b: Extractor[R, B],
c: Extractor[R, C],
d: Extractor[R, D],
e: Extractor[R, E],
f: Extractor[R, F],
g: Extractor[R, G],
h: Extractor[R, H],
) -> Extractor[R, M]
Combine several extractors into one that builds a typed value.
The escape hatch from a handler's extractor-arity ceiling, and the way to
parse a group of inputs into one model: make is the model's constructor (or
any factory) and each extractor supplies one positional argument to it, in
order, with the types tied so a mismatch is a mypy error. The constituents'
OpenAPI fragments (query/header/body) are carried through; path parameters
still appear in the route pattern, so their schema comes from there.
This reuses the existing tokens rather than re-reading the request: pass the
same path_param/query_param values you would otherwise hand the handler,
plus the type that assembles them.
make is called positionally, which a frozen dataclass or NamedTuple
constructor accepts directly. For a pydantic model (whose __init__ is
keyword-only, and whose validators you want to run), pass a small factory
that constructs it by keyword: into(lambda a, b: M(x=a, y=b), ea, eb). A
validator that rejects raises ValidationError, which the router's exception
handlers map like any other parse failure.
Adapt a single-value parse into the tuple-taking form query_param and
header_param expect, requiring the value to appear exactly once.
Use it for a singleton field that must be present once: it raises ValueError
when the value is absent or repeated (a duplicated singleton is a protocol
violation, RFC 9110 §5.3) and otherwise applies parse to the sole value. For a
genuinely list-valued field, skip this and let parse take every value.
Like once, but for a field that may appear zero or one times.
Returns None when the value is absent and parse(value) when it appears once;
a repeated value still raises ValueError (a duplicated singleton is a protocol
violation, RFC 9110 §5.3). Use it for an optional singleton field, and once
when the field is required.
path_param(
name: str, converter: Converter[V]
) -> Extractor[RequestHead, V]
A typed path parameter: one value that is both a pattern segment and a read.
The same converter the router matches the segment with also fixes the type
V the handler receives, so there is no second place to keep in sync: drop
this extractor into the route pattern (("todos", path_param("id", INT)))
and into the handler's argument list, and the name, converter, schema, and
type are all declared exactly once. The read casts the value the router's
walk already parsed with this very converter, so the cast is sound.
query_param(
name: str,
parse: Callable[[tuple[str, ...]], V],
*,
schema: SchemaRef,
required: bool = False,
) -> Extractor[RequestHead, V]
Parse a query parameter into V, given all of its raw values.
parse receives the (possibly empty, possibly repeated) values for name as an
immutable tuple and decides what their absence and multiplicity mean, returning
V or raising a ValueError to reject, which becomes an ExtractionError
naming this name. The schema is this value's OpenAPI contribution.
websocket_scope() -> Extractor[
WebsocketRequestHead, WebsocketScope
]
Hand a websocket handler the unparsed WebsocketScope.
The websocket sibling of http_scope(): its context is WebsocketRequestHead,
so it reads the narrowed scope with no runtime check, and use on an HTTP route
is a static type error.
handle(
*,
fn: Callable[[T], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
/,
*,
fn: Callable[[T, A], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
/,
*,
fn: Callable[[T, A, B], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
/,
*,
fn: Callable[[T, A, B, C], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
/,
*,
fn: Callable[[T, A, B, C, D], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
e: Extractor[BufferedRequest, E],
/,
*,
fn: Callable[[T, A, B, C, D, E], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
e: Extractor[BufferedRequest, E],
f: Extractor[BufferedRequest, F],
/,
*,
fn: Callable[[T, A, B, C, D, E, F], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
e: Extractor[BufferedRequest, E],
f: Extractor[BufferedRequest, F],
g: Extractor[BufferedRequest, G],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, G], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
e: Extractor[BufferedRequest, E],
f: Extractor[BufferedRequest, F],
g: Extractor[BufferedRequest, G],
h: Extractor[BufferedRequest, H],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, G, H], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
e: Extractor[BufferedRequest, E],
f: Extractor[BufferedRequest, F],
g: Extractor[BufferedRequest, G],
h: Extractor[BufferedRequest, H],
j: Extractor[BufferedRequest, J],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, G, H, J], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[BufferedRequest, A],
b: Extractor[BufferedRequest, B],
c: Extractor[BufferedRequest, C],
d: Extractor[BufferedRequest, D],
e: Extractor[BufferedRequest, E],
f: Extractor[BufferedRequest, F],
g: Extractor[BufferedRequest, G],
h: Extractor[BufferedRequest, H],
j: Extractor[BufferedRequest, J],
k: Extractor[BufferedRequest, K],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, G, H, J, K],
Returned,
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
*extractors: Extractor[BufferedRequest, object],
fn: Callable[..., Returned],
summary: str = "",
responses: Mapping[int, ResponseSpec] | None = None,
) -> HttpEndpoint[T]
Build a self-describing endpoint from typed extractors and a handler.
Each extractor is a typed piece of the request; the overloads tie the
extractors' types to fn's parameters, so a path_param(..., INT) paired
with an fn that expects a str is a mypy error, not a runtime surprise.
At dispatch the input body is buffered once, a BufferedRequest is built,
every extractor runs (raising to reject, mapped by the router's exception
handlers), and fn is called with the typed values. The handler is always
async; the output is free: an async def that resolves to a Response,
or an async def ... yield that streams Outbound events, and _emit relays
whichever. The endpoint also answers describe(), recovering its
query/header/body OpenAPI from the same extractors.
handle is the lower-level builder; reach for the @get/@post/... method
decorators to co-locate the route with its handler.
handle_stream(
*,
fn: Callable[[T, Stream[Inbound]], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
/,
*,
fn: Callable[[T, A, Stream[Inbound]], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
/,
*,
fn: Callable[[T, A, B, Stream[Inbound]], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
/,
*,
fn: Callable[
[T, A, B, C, Stream[Inbound]], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
/,
*,
fn: Callable[
[T, A, B, C, D, Stream[Inbound]], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
e: Extractor[HttpRequestHead, E],
/,
*,
fn: Callable[
[T, A, B, C, D, E, Stream[Inbound]], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
e: Extractor[HttpRequestHead, E],
f: Extractor[HttpRequestHead, F],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, Stream[Inbound]],
Returned,
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
e: Extractor[HttpRequestHead, E],
f: Extractor[HttpRequestHead, F],
g: Extractor[HttpRequestHead, G],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, G, Stream[Inbound]],
Returned,
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
e: Extractor[HttpRequestHead, E],
f: Extractor[HttpRequestHead, F],
g: Extractor[HttpRequestHead, G],
h: Extractor[HttpRequestHead, H],
/,
*,
fn: Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
Stream[Inbound],
],
Returned,
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
e: Extractor[HttpRequestHead, E],
f: Extractor[HttpRequestHead, F],
g: Extractor[HttpRequestHead, G],
h: Extractor[HttpRequestHead, H],
j: Extractor[HttpRequestHead, J],
/,
*,
fn: Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
J,
Stream[Inbound],
],
Returned,
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
a: Extractor[HttpRequestHead, A],
b: Extractor[HttpRequestHead, B],
c: Extractor[HttpRequestHead, C],
d: Extractor[HttpRequestHead, D],
e: Extractor[HttpRequestHead, E],
f: Extractor[HttpRequestHead, F],
g: Extractor[HttpRequestHead, G],
h: Extractor[HttpRequestHead, H],
j: Extractor[HttpRequestHead, J],
k: Extractor[HttpRequestHead, K],
/,
*,
fn: Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
J,
K,
Stream[Inbound],
],
Returned,
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
request_body: Body | None = ...,
) -> HttpEndpoint[T]
handle_stream(
*extractors: Extractor[HttpRequestHead, object],
fn: Callable[..., Returned],
summary: str = "",
responses: Mapping[int, ResponseSpec] | None = None,
request_body: Body | None = None,
) -> HttpEndpoint[T]
Build an endpoint whose handler reads the inbound stream live.
The streaming-input sibling of handle. Where handle buffers the request
body before the handler runs (so a body extractor can read it), this leaves
the inbound stream untouched and hands it to the handler as a trailing
Stream[Inbound] argument: the handler is the processor, taking the state,
the typed extractor values, and the live stream, reading it as events arrive
(a streaming upload, a long poll, a loop driven by request chunks). The
extractors are scope-only (path_param/query_param/header_param/
http_scope, whose context is the streaming route's HttpRequestHead); a body
extractor is a static type error, since its BufferedRequest context is exactly
the buffering a streaming route avoids. The output is free, exactly as in
handle: yield Outbound to stream the response, or return (or await) a
Response to buffer it.
Reach for the @get.stream/@post.stream/... method decorators to co-locate
the streaming route with its handler.
ws(
pattern: Pattern,
) -> Callable[
[
Callable[
[T, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern, a: Extractor[WebsocketRequestHead, A]
) -> Callable[
[
Callable[
[T, A, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
) -> Callable[
[
Callable[
[T, A, B, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
) -> Callable[
[
Callable[
[T, A, B, C, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
) -> Callable[
[
Callable[
[T, A, B, C, D, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
e: Extractor[WebsocketRequestHead, E],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
e: Extractor[WebsocketRequestHead, E],
f: Extractor[WebsocketRequestHead, F],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
e: Extractor[WebsocketRequestHead, E],
f: Extractor[WebsocketRequestHead, F],
g: Extractor[WebsocketRequestHead, G],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
e: Extractor[WebsocketRequestHead, E],
f: Extractor[WebsocketRequestHead, F],
g: Extractor[WebsocketRequestHead, G],
h: Extractor[WebsocketRequestHead, H],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
e: Extractor[WebsocketRequestHead, E],
f: Extractor[WebsocketRequestHead, F],
g: Extractor[WebsocketRequestHead, G],
h: Extractor[WebsocketRequestHead, H],
j: Extractor[WebsocketRequestHead, J],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
J,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[WebsocketRequestHead, A],
b: Extractor[WebsocketRequestHead, B],
c: Extractor[WebsocketRequestHead, C],
d: Extractor[WebsocketRequestHead, D],
e: Extractor[WebsocketRequestHead, E],
f: Extractor[WebsocketRequestHead, F],
g: Extractor[WebsocketRequestHead, G],
h: Extractor[WebsocketRequestHead, H],
j: Extractor[WebsocketRequestHead, J],
k: Extractor[WebsocketRequestHead, K],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
J,
K,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
*extractors: Extractor[WebsocketRequestHead, object],
) -> Callable[
[Callable[..., WebsocketReturned]], WebsocketRoute[T]
]
The websocket sibling of @get/@post, tying extractors to a handler.
@ws(t"/feed/{room}", room, since) co-locates the route with the handler and
ties each extractor's type to its parameters, just like @get. The handler
is the frame processor (the same move as @post.stream): it takes the live
inbound frames as a trailing Stream[WebsocketInbound] argument and yields
WebsocketOutbound, rather than returning a processor. There is no body to
buffer: path_param, query_param, and header_param read the handshake, and a
body (or http_scope) token is a static type error, its context not the
WebsocketRequestHead a websocket route provides. Returns a WebsocketRoute to
pass to a WebsocketRouter.
describe(
spec: RouteSpec,
) -> Callable[
[Endpoint[T, HttpScope, HttpHandler]],
Endpoint[T, HttpScope, HttpHandler],
]
Attach a RouteSpec to an endpoint, making it self-describing.
The same value the handler is built around (its body/response types) becomes
its OpenAPI contribution: one declaration, two consumers. Reads as a
decorator above buffered, so the endpoint stays a plain callable that also
answers describe().
Split a request target into its segments.
Leading and trailing slashes are stripped, so / is the empty tuple and a
trailing slash never produces an empty segment: /users and /users/ both
split to ("users",). Matching is therefore trailing-slash insensitive,
because targets and the literal parts of patterns are split by this same
function.
buffered(
make: Callable[[T, Match[HttpScope], bytes], Response],
) -> Endpoint[T, HttpScope, HttpHandler]
Adapt a body-reading (state, match, body) -> Response into an Endpoint.
The web-flavored sibling of without_asgi.routing.buffered: it hands the
handler the Match (the scope plus the router's already-parsed path
parameters) rather than the bare scope, so a handler reads match.params
and match.scope without re-parsing the target. Reads the whole request
body, then runs make once and emits the single Response. Usable as a
decorator.
delegate(
prefix: str, target: HttpRouter[T]
) -> Delegate[T]
Mount an opaque HTTP app at prefix (see Delegate).
Bake a literal prefix (and optional per-route middleware) into HTTP routes.
Returns a transform that rebases each Route/Delegate you hand it: the prefix
is prepended to the path and the middleware wrapped onto each endpoint. The
result is a plain route whose segments already include the prefix, so there is
no Mount wrapper in the router, matching and OpenAPI see the full path, and
reverse routing (url_for) needs no router. Store and reuse it, apply it to
many routes at once, or use it as a decorator on one:
api = mount("/api", require_auth) # a reusable mount point
routes = api(list_users, create_user) # -> a tuple of rebased routes
@mount("/api") # or as a decorator on one route
@get(t"/users/{uid}", uid)
async def show_user(...): ...
Nesting composes (mount("/api")(mount("/v1")(r)) -> /api/v1/...). For a
sub-app whose routes you cannot see, use delegate(...) instead: an opaque app
cannot have a prefix baked in, so it stays a black box handed the trimmed scope.
route(
pattern: Pattern,
*,
get: HttpEndpoint[T] | None = None,
head: HttpEndpoint[T] | None = None,
post: HttpEndpoint[T] | None = None,
put: HttpEndpoint[T] | None = None,
patch: HttpEndpoint[T] | None = None,
delete: HttpEndpoint[T] | None = None,
options: HttpEndpoint[T] | None = None,
) -> Route[T]
Build a Route, one endpoint per method keyword.
url_for(
route: Reversible,
values: Mapping[str, object] = _NO_VALUES,
) -> str
Reverse a route to a concrete path: the inverse of the trie walk, as a pure function.
url_for(route, values) fills the route's path parameters from values and
renders the path it would match at, so a handler or template links to a route by
its value rather than hand-assembling a string that drifts when the path
changes. Because mount bakes any prefix into the route, the route's segments
are its full path: reversing needs no router, holds no hidden prefix for another
router to be ignorant of, and works the same whether the route came from
@get, mount(...), or a third-party package. A handler links by referencing
the route value (immutable), never the assembled router.
Each value is rendered and fed back through its segment's converter to prove it
would parse straight back (parse, don't validate, in reverse): a value the
converter would reject, one that does not round-trip, or a single-segment value
containing / raises, as does a missing or unknown parameter. A catch_all
segment is the one place / is allowed.
with_middleware(
endpoint: Endpoint[T, S, H],
*middleware: Middleware[T, H, S],
) -> Endpoint[T, S, H]
Scope middleware to one endpoint instead of the whole router.
The router-wide middleware runs on every dispatch; this applies the same
Middleware vocabulary to a single route (or an opaque delegate target). An
Endpoint builds the handler and a Middleware is (handler, T, S) -> handler,
so this is just composition: build the handler, then run the middleware over it
with the request's scope. First argument is outermost, matching stack. Use it per
method, e.g. route("/admin", get=with_middleware(list_admins, require_auth));
for a whole prefix, hand the middleware to mount(...).
ws_delegate(
prefix: str, target: WebsocketRouter[T]
) -> WebsocketDelegate[T]
Mount an opaque WebSocket app at prefix (see WebsocketDelegate).
The WebSocket sibling of mount: bake a prefix (and middleware) into WebSocket routes.
ws_route(
pattern: Pattern, endpoint: WebsocketEndpoint[T]
) -> WebsocketRoute[T]
Build a WebsocketRoute.