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
¶
Extractor(
extract: Callable[[Request], _V_co],
query: tuple[QueryParam, ...] = (),
headers: tuple[HeaderParam, ...] = (),
request_body: Body | None = None,
path: PathSpec | None = None,
)
Bases: Generic[_V_co]
A typed piece of a request, paired with the OpenAPI it contributes.
extract is a pure Request -> V 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 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
¶
The parsed-once context an extractor reads from.
params holds the path parameters the router already parsed during the trie
walk (typed values, stored as object); body is the fully-buffered request
body (empty for a websocket handshake, which has none). One value, built once
per connection, fed to every extractor a handler declares.
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.
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
¶
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.
dataclass
¶
Mount(prefix: str, target: Router[T] | HttpRouter[T])
A sub-application mounted at a literal-string prefix.
target is either a without-web Router (whose routes are grafted into
this router's trie, so matching and OpenAPI see straight through with the
prefix prepended) or an opaque HttpRouter (a BYO router or another app),
which is handed the prefix-trimmed scope and treated as a black box.
The prefix is a plain str, hence a literal path: it cannot carry a path
parameter, since that would need a Template (as route patterns use) and the
type does not allow one here. The opaque-mount trim strips a fixed string
(root_path semantics), so there is nowhere for a parameter to bind anyway.
dataclass
¶
Route(
pattern: Pattern,
methods: Mapping[str, HttpEndpoint[T]],
)
A path pattern bound to one endpoint per HTTP method.
The method-decorator form (@get(...)) produces a single-method Route; the
Router merges Routes that share a pattern into one method map, so the
405-vs-404 split still falls out of the trie.
dataclass
¶
Router(
routes: tuple[Route[T] | Mount[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 mounting from the tree's shape.
dataclass
¶
WebsocketRoute(
pattern: Pattern, endpoint: WebsocketEndpoint[T]
)
A path pattern bound to one WebSocket endpoint.
dataclass
¶
WebsocketRouter(
routes: tuple[WebsocketRoute[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=...).
class-attribute
instance-attribute
¶
class-attribute
instance-attribute
¶
tree: Node[WebsocketEndpoint[T]] = field(
init=False, repr=False, compare=False
)
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[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 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[[list[bytes]], V],
*,
schema: SchemaRef,
required: bool = False,
) -> Extractor[V]
Parse a request header into V, given all of its raw values.
Header names are matched case-insensitively (ASGI lower-cases them); parse
receives every value sent under name, in order, and returns V or raises.
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. Using it off
an HTTP route is the invariant; the assert turns misuse on a websocket route
into a loud error rather than a confusing AttributeError downstream.
into(
make: Callable[[A, B, C], M],
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
) -> Extractor[M]
into(
make: Callable[[A, B, C, D], M],
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
) -> Extractor[M]
into(
make: Callable[[A, B, C, D, E], M],
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
) -> Extractor[M]
into(
make: Callable[[A, B, C, D, E, F], M],
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
) -> Extractor[M]
into(
make: Callable[[A, B, C, D, E, F, G], M],
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
) -> Extractor[M]
into(
make: Callable[[A, B, C, D, E, F, G, H], M],
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
) -> Extractor[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.
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[[list[str]], V],
*,
schema: SchemaRef,
required: bool = False,
) -> Extractor[V]
Parse a query parameter into V, given all of its raw values.
parse receives the (possibly empty, possibly repeated) values for name
and decides what their absence and multiplicity mean, returning V or
raising to reject. The schema is this value's OpenAPI contribution.
websocket_scope() -> Extractor[WebsocketScope]
Hand a websocket handler the unparsed WebsocketScope.
The websocket sibling of http_scope(); the assert guards against using it on
an HTTP route.
handle(
*,
fn: Callable[[T], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
/,
*,
fn: Callable[[T, A], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
/,
*,
fn: Callable[[T, A, B], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
/,
*,
fn: Callable[[T, A, B, C], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
/,
*,
fn: Callable[[T, A, B, C, D], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
/,
*,
fn: Callable[[T, A, B, C, D, E], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
/,
*,
fn: Callable[[T, A, B, C, D, E, F], Returned],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
/,
*,
fn: Callable[
[T, A, B, C, D, E, F, G], Returned
],
summary: str = ...,
responses: Mapping[int, ResponseSpec] | None = ...,
) -> HttpEndpoint[T]
handle(
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
j: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
j: Extractor[J],
k: Extractor[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[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 Request 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[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[A],
b: Extractor[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[A],
b: Extractor[B],
c: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
j: Extractor[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[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
j: Extractor[J],
k: Extractor[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[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); a body extractor is rejected, since buffering the body is
exactly what 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[A]
) -> Callable[
[
Callable[
[T, A, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern, a: Extractor[A], b: Extractor[B]
) -> Callable[
[
Callable[
[T, A, B, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
) -> Callable[
[
Callable[
[T, A, B, C, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
) -> Callable[
[
Callable[
[T, A, B, C, D, Stream[WebsocketInbound]],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
j: Extractor[J],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
J,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern,
a: Extractor[A],
b: Extractor[B],
c: Extractor[C],
d: Extractor[D],
e: Extractor[E],
f: Extractor[F],
g: Extractor[G],
h: Extractor[H],
j: Extractor[J],
k: Extractor[K],
) -> Callable[
[
Callable[
[
T,
A,
B,
C,
D,
E,
F,
G,
H,
J,
K,
Stream[WebsocketInbound],
],
WebsocketReturned,
]
],
WebsocketRoute[T],
]
ws(
pattern: Pattern, *extractors: Extractor[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 (a
body extractor is rejected, since a handshake carries none). 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.
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.
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 mount 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 mounted subtree, give the mounted Router its own middleware.
ws_route(
pattern: Pattern, endpoint: WebsocketEndpoint[T]
) -> WebsocketRoute[T]
Build a WebsocketRoute.