`blackbull`¶

`blackbull` ¶

BlackBull — async ASGI 3.0 web framework.

Early Alpha — API may break between MINOR versions; see KNOWN_LIMITATIONS.md before building production-shape work on top.

Public API exports — __all__ at the foot of this module is the complete and authoritative list; the notes below cover the ones worth a sentence:

BlackBull: the main application object; wraps routing, middleware, and lifespan hooks.
AppConfig: declarative, immutable holder for the startup settings run() accepts (port, TLS, workers, …).
serve: synchronous entry point that runs any ASGI 3.0 callable (also used by the blackbull console script).
Response, JSONResponse, RedirectResponse, StreamingResponse, EventSourceResponse, WebSocketResponse: response helpers.
RouteInfo: immutable (method, path, name) snapshot returned by app.get_routes().
QUERY: the HTTP QUERY method (RFC 10008) as a plain string — http.HTTPMethod lacks the member until Python ≥3.16.
UnprocessableQuery: raise from a QUERY handler for 422 when the (accepted) request media type carries a semantically unprocessable query (RFC 10008).
Headers: case-insensitive, ordered, multi-valued HTTP header store.
Connection: the typed internal request representation; the handler context object exposing method/path/raw_path/query_string/headers/cookies/query/query_list/path_params/state and body()/json()/text()/form(). The ASGI scope is a derived view (Connection.as_scope()).
Request: deprecated alias of Connection. Accessing blackbull.Request emits a DeprecationWarning; replace request: Request handler params with conn: Connection (identical API). Removal no earlier than 2027-08-01.
WebSocket: the high-level WebSocket handler object — await ws.accept(), async for message in ws, ws.send_text()/send_bytes()/send_json(), await ws.close(). Declare async def handler(ws: WebSocket) on a Scheme.websocket route to receive it; the raw (conn, receive, send) form keeps working unchanged.
WebSocketDisconnect: raised by ws.receive() when the peer closes; carries code and reason. async for ends the loop instead of raising.
Depends: per-request provider injection for simplified handlers (async-generator providers get teardown after the response is sent).
cookie_header: builds a Set-Cookie header tuple.
read_body: reads and buffers the full request body from the ASGI receive channel.
read_json: reads the body and parses it as JSON (None on empty/invalid).
read_text: reads the body and decodes it as text.
parse_cookies: parses the Cookie header into a plain dict.
CORS: adds Access-Control-* headers; handles preflight OPTIONS requests.
as_middleware: decorator that marks an async function or class as middleware; normalises send so inner wrappers see only ASGI event dicts.
TrustedProxy: rewrites scope['client'] / scope['scheme'] from proxy headers.

Importing this package loads the server stack (blackbull.server.*) as a side effect: blackbull.app imports RawBinding from blackbull.server.protocol_registry, and blackbull/server/__init__.py re-exports ASGIServer, so the whole of it is resolved before BlackBull itself is bound. Use ASGIServer from blackbull.server to embed BlackBull's own server; otherwise pass the BlackBull instance to any external ASGI server (uvicorn, hypercorn, granian, …) since BlackBull.__call__ is ASGI 3.0 compliant.

`QUERY = 'QUERY'` `module-attribute` ¶

The HTTP QUERY method (RFC 10008) — safe, idempotent, cacheable, with a request body.

http.HTTPMethod has no QUERY member (RFC 10008 postdates the 3.15 feature freeze; earliest stdlib arrival is 3.16), so BlackBull exports the method as a plain string usable anywhere a method is accepted::

from blackbull import QUERY

@app.route(path='/search', methods=[QUERY])
async def search(body: bytes): ...

Because http.HTTPMethod is a StrEnum, this string stays equal- and hash-compatible with a future HTTPMethod.QUERY member — routes registered with it need no migration.

`AppConfig` `dataclass` ¶

Immutable, declarative startup configuration for a BlackBull app.

Every field corresponds one-to-one with a keyword argument of :func:blackbull.serve / :meth:blackbull.BlackBull.run. Fields left at their sentinel default (None, or 0 for port, or False for reload) defer to serve's own built-in default unless an explicit run(...) argument overrides them.

`BlackBull` ¶

`add_extension(ext)` ¶

Register an extension and return it (for decorator chaining).

ext is any object exposing init_app(app) — a :class:~blackbull.extension.Extension subclass, or a legacy duck-typed extension. init_app is called immediately to wire the extension's routes / middleware / protocol handlers / events through the public app.* API. Optional async startup(app) / shutdown(app) methods are wired into the app_startup / app_shutdown lifespan events.

This is the only extension- or protocol-registration entry point on the core class; protocol support (e.g. an MQTT broker) is added by passing the relevant extension here, never by editing this class::

from blackbull.mqtt import MQTTExtension, Message

mqtt = app.add_extension(MQTTExtension(port=1883))

@mqtt.on_message(topic='sensors/+/temperature')
async def on_temp(msg: Message):
    ...

Returns ext so it can be captured and configured further.

`drain_events(timeout=5.0)` `async` ¶

Wait for detached (@app.on) observers to finish. Returns success.

Two of the three hook kinds need no seam: @app.intercept and @app.on(..., blocking=True) are awaited before a request returns, so their effects are already visible. The third is detached on purpose, so asserting its side-effect straight after a request is a race — this is how a test waits instead of sleeping.

False means timeout expired with work still outstanding. Nothing is cancelled; call again with a longer budget.

`enable_grpc(registry)` ¶

Serve unary gRPC calls from registry over the HTTP/2 layer.

gRPC is HTTP/2 with content-type: application/grpc; once enabled, such requests are dispatched to the registry's handlers (returning grpc-status trailers) instead of the HTTP router, while REST and WebSocket traffic on the same port is unaffected.

registry is a :class:blackbull.grpc.GrpcServiceRegistry. gRPC requires HTTP/2, so run the app with TLS+ALPN (or h2c) for real clients. Protobuf is not pulled in — handlers exchange raw message bytes; see blackbull.grpc for the handler contract.

`enable_openapi(*, title='BlackBull API', version='0.1.0', description=None, spec_path='/openapi.json', docs_path='/docs')` ¶

Auto-publish an OpenAPI 3.1 spec and Swagger UI for the app.

Registers two routes:

spec_path (default /openapi.json) returns the spec as JSON. The spec is regenerated on every request so new routes added after this call are reflected.
docs_path (default /docs) returns an HTML page hosting Swagger UI pointed at spec_path. Pass docs_path=None to skip the UI route and serve only the JSON spec.

Call once, after the rest of the app's routes have been registered.

This is a thin convenience wrapper around OpenAPIExtension, which is the reference implementation of the init_app(app) extension convention (see the Extensions guide). External callers may also instantiate the extension class directly when they want to keep a handle on it after registration::

from blackbull.openapi import OpenAPIExtension
ext = OpenAPIExtension(app, title='My API')
assert app.extensions['openapi'] is ext

`get_routes()` ¶

Return a snapshot of all registered routes.

Each entry is a :class:~blackbull.router.RouteInfo named tuple (method, path, name). Routes are returned in registration order, one entry per HTTP method. The list is a shallow copy and may be freely sorted, filtered, or mutated without affecting the live router.

This is the public, stable alternative to reaching into the internal app._router._route_info attribute — use it for dashboards, OpenAPI generators, admin panels, and debug endpoints.

`group(middlewares=[])` ¶

Return a RouteGroup that prepends middlewares to every route.

`intercept(event_name)` ¶

Decorate a handler to intercept event_name synchronously.

The handler is awaited in registration order when the event fires. Exceptions propagate to the emitter and abort subsequent interceptors registered for the same event.

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event to intercept (e.g. `'app_startup'`).	required

Returns:

Type	Description
`Callable[[EventHandler], EventHandler]`	A decorator that registers the wrapped coroutine and returns it
`Callable[[EventHandler], EventHandler]`	unchanged.

Example

@app.intercept('app_startup')
async def handler(event: Event):
    await setup()

`on(event_name, *, blocking=False)` ¶

Decorate a handler to observe event_name.

With blocking=False (the default) the handler is scheduled as an independent asyncio.Task each time the event fires — fire-and-forget, never delaying the emitter. Use it for telemetry that must not add latency to the hot path.

With blocking=True the handler is awaited in registration order before the emit completes, so a side effect is guaranteed to finish within the event's lifetime. Use it for resource cleanup keyed to an event's completion — most notably scope_completed (close a per-request DB session, delete a temp file).

Either way the handler's exceptions are isolated: they are caught and logged and never propagate to the emitter or affect other handlers. (For a hook that may affect the request, use :meth:intercept.)

Parameters:

Name	Type	Description	Default
`event_name`	`str`	Name of the event to observe (e.g. `'scope_completed'`).	required
`blocking`	`bool`	Await the handler before emit returns (default `False`).	`False`

Returns:

Type	Description
`Callable[[EventHandler], EventHandler]`	A decorator that registers the wrapped coroutine and returns it
`Callable[[EventHandler], EventHandler]`	unchanged.

Example

@app.on('request_completed')            # detached telemetry
async def log_it(event: Event):
    metrics.record(event.detail)

@app.on('scope_completed', blocking=True)   # awaited cleanup
async def close_session(event: Event):
    session = event.detail['conn'].get('state', {}).get('db')
    if session is not None:
        await session.close()

`on_error(key)` ¶

Register a custom error handler for an HTTPStatus or exception class.

key may be an :class:HTTPStatus, a plain int status code (coerced to HTTPStatus), or an exception class.

Usage::

@app.on_error(HTTPStatus.FORBIDDEN)
async def handle_403(conn, receive, send):
    ...

@app.on_error(403)            # int shorthand
async def handle_403(conn, receive, send):
    ...

@app.on_error(ValueError)
async def handle_value_error(conn, receive, send):
    ...

The handler receives (conn, receive, send). conn.state contains: - 'error_status' : HTTPStatus - 'error_exception' : exception instance (when triggered by an exception) - 'allowed_methods' : allowed method names (for 405)

`on_shutdown(fn)` ¶

Register a zero-argument coroutine to run at lifespan shutdown.

The handler is wrapped in an adapter and registered as an 'app_shutdown' interception handler so it runs before the ASGI server receives the lifespan.shutdown.complete acknowledgement. Shutdown handlers run in registration order; an exception aborts the remaining handlers.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[], Awaitable[None]]`	Async callable that takes no arguments.	required

Returns:

Type	Description
`Callable[[], Awaitable[None]]`	fn unchanged, so the decorator can be stacked or the function
`Callable[[], Awaitable[None]]`	used normally after registration.

Example

@app.on_shutdown
async def close_db():
    await db.disconnect()

`on_startup(fn)` ¶

Register a zero-argument coroutine to run at lifespan startup.

The handler is wrapped in an adapter and registered as an 'app_startup' interception handler so it runs before the ASGI server receives the lifespan.startup.complete acknowledgement. Startup handlers run in registration order; an exception aborts the remaining handlers and prevents the completion event from being sent.

Parameters:

Name	Type	Description	Default
`fn`	`Callable[[], Awaitable[None]]`	Async callable that takes no arguments.	required

Returns:

Type	Description
`Callable[[], Awaitable[None]]`	fn unchanged, so the decorator can be stacked or the function
`Callable[[], Awaitable[None]]`	used normally after registration.

Example

@app.on_startup
async def open_db():
    await db.connect()

`on_warmup(fn)` ¶

Register a coroutine to warm the app before it binds or forks.

Unlike :meth:on_startup (which runs inside each worker's lifespan, after fork() and after the listening socket already exists), an on_warmup hook runs once, in the master, before the socket is created and before workers are forked. Forked workers then inherit the warmed heap via copy-on-write (PEP 659 specialization survives fork(); the framework calls gc.collect() + gc.freeze() after warm-up to keep those pages shared). In single-worker mode the one process is warmed before it binds.

Hooks receive the app and must do pure warming only — drive hot code paths, prime codecs/TLS — and acquire no per-worker resources (DB pools, sockets, live connections); those belong in :meth:on_startup, which runs per worker. Use :meth:warm_request to exercise the ASGI dispatch/handler path in-process, and :func:blackbull.server.warmup.warm_tls to prime the TLS handshake.

Warm-up is best-effort: a hook's exception is logged and swallowed (the master degrades to a cold start), and total warm-up time is capped by BB_WARMUP_BUDGET_S (default 60 s). Multiple hooks run in registration order.

Example::

@app.on_warmup
async def warm(app):
    from blackbull import Connection, Headers
    conn = Connection(
        method='POST', path='/rpc', raw_path=b'/rpc',
        headers=Headers([(b'content-type', b'application/grpc')]))
    await app.warm_request(conn, body=req_bytes, n=2000)

`raw_handler(name, *, port=None, detector=None, tls=False, stateful=True)` ¶

Decorator form of :meth:register_protocol_handler.

::

@app.raw_handler('echo', port=9000)
async def echo(reader, writer, ctx):
    while data := await reader.read(1024):
        await writer.write(data)

`register_converter(type_, converter=None)` ¶

Teach simplified handlers to return values of a custom type_.

A simplified handler may already return str, bytes, dict, list, a dataclass, a Response, or None. Register a converter to extend that set — e.g. so a handler can return my_orm_object and have it serialised automatically. The converter receives the returned value and must return a natively supported sendable (a Response, str/bytes, None, or a JSON-able dict/list/dataclass).

The registry is empty by default, so registering nothing costs nothing: the coercion fast path never consults it for the built-in shapes.

Direct form::

app.register_converter(MyOrmObject, lambda o: o.to_dict())

Decorator form (omit converter)::

@app.register_converter(MyOrmObject)
def _(o):
    return o.to_dict()

Converters registered after a route are still honoured — the registry is shared with every adapted handler by reference.

`register_protocol_handler(name, handler, *, detector=None, port=None, tls=False, stateful=True)` ¶

Register a handler for a non-ASGI (raw) protocol.

The handler is an async callable (reader, writer, ctx) -> None that owns the connection for its whole lifetime. When port is set, the server binds an additional listening socket on it; connections there skip HTTP detection and go straight to handler.

Parameters:

Name	Type	Description	Default
`name`	`str`	Protocol name (e.g. `'echo'`, `'mqtt'`); must be unique.	required
`handler`	`Callable[..., Awaitable[None]]`	Async `(reader, writer, ctx)` coroutine.	required
`detector`	`object \| None`	First-byte sniffing on the shared HTTP port, so this protocol can be reached there as well as on its own port.	`None`
`port`	`int \| None`	Dedicated listening port for this protocol.	`None`
`tls`	`bool`	Serve this port through the server's TLS machinery (e.g. `mqtts://`). Requires the server to be configured with a certificate; startup fails fast otherwise.	`False`
`stateful`	`bool`	Whether an exchange depends on what an earlier one left behind — true by default. A stateful protocol is served by one worker, so with `workers > 1` it is reached on its own port only; the shared port would answer from whichever worker accepted. A stateful protocol with no dedicated port cannot be given one owner at all and is refused before the workers fork. Pass `False` for a protocol that keeps nothing between exchanges.	`True`

Returns:

Type	Description
`RawBinding`	The registered binding.

`route(methods=[HTTPMethod.GET], path='/', scheme=Scheme.http, functions=[], middlewares=[], name=None, accept_query=None)` ¶

Register a route handler, optionally wrapping it in middlewares.

accept_query (RFC 10008) names the request media types the route accepts — it is the value of the Accept-Query response header, not a switch that enables the QUERY method. (A method is accepted purely by listing it in methods; QUERY is no different from GET there.) It is meaningful for the QUERY method, which carries a request body. When set, the route's responses carry an Accept-Query header (an RFC 9651 Structured Field list of those media types), and QUERY requests are Content-Type-enforced: a missing media type is answered 400, an unaccepted one 415 (with the Accept-Query header so the client can correct). A handler may raise :class:~blackbull.UnprocessableQuery for 422 on a well-formed but unprocessable query. Enforcement applies only to QUERY requests; other methods on the same route still receive the Accept-Query header.

`run(certfile=None, keyfile=None, port=None, unix_path=None, inherited_fd=None, listeners=None, workers=None, max_connections=None, stream_queue_depth=None, ws_queue_depth=None, reload=None, reload_paths=None)` ¶

Run the app under BlackBull's own server (single- or multi-worker).

This is the synchronous, fire-and-forget entry point — callers write app.run(port=8000), not asyncio.run(app.run(...)). For workers > 1 or reload=True the master pre-binds sockets, forks workers, and blocks until SIGTERM / SIGINT.

Each argument left unset (None) is resolved, highest precedence first: the explicit argument → a BLACKBULL_* environment variable (BLACKBULL_PORT / CERT / KEY / UNIX_PATH / RELOAD) → a .env file in the working directory (needs the [dotenv] extra) → the bound :class:~blackbull.AppConfig → :func:blackbull.serve's own default. Server-tuning knobs (workers, max_connections, queue depths) keep their BB_* environment variables; BLACKBULL_* is the deployment namespace. The provenance of each non-default deploy setting is logged once at startup on the blackbull.config logger::

app = BlackBull(config=AppConfig(port=8443, certfile='c.pem',
                                 keyfile='k.pem'))
app.run()              # binds 8443 with TLS from the config
app.run(port=9000)     # explicit arg overrides the config's 8443
# BLACKBULL_PORT=9000 python app.py   # env overrides the config too

For embedded use under an existing event loop, or for pre-binding a socket before forking a test subprocess, instantiate :class:blackbull.server.ASGIServer directly. Any external ASGI server (uvicorn / hypercorn / granian / …) can drive the :class:BlackBull instance via its ASGI 3.0 __call__.

Example::

app.run(port=8000)
app.run(port=8443, certfile='cert.pem', keyfile='key.pem', workers=4)
app.run(port=8443, certfile='cert.pem', keyfile='key.pem', reload=True)
app.run(unix_path='/run/blackbull.sock')

`static(url_prefix, root_dir, *, cache=False, index=None, conditional=True)` ¶

Serve static files from root_dir under url_prefix as a route.

cache (default False): when True, file bodies are held in-memory for fast cache-hit serving. Useful for standalone deployments where BlackBull terminates static traffic directly. Most production deployments place nginx / a CDN in front of the framework for static traffic and don't need the in-process cache; the default off is calibrated to that majority. See docs/guide/static-files.md for the full discussion.

index (default None — off): a filename (e.g. 'index.html') served when a request resolves to a directory.

conditional (default True): emit ETag / Last-Modified validators and answer If-None-Match / If-Modified-Since with a 304. Set False to disable conditional responses.

Registered as a route, not global middleware, so a request that is not for a static path never enters this code: it resolves in the router's exact-match dict and never reaches the parametrised scan. This is also what every peer framework does — Starlette, Sanic, aiohttp, Flask and Django all mount static as a route.

`to_asgi()` ¶

Return the ASGI 3.0 callable to hand to an external host (uvicorn …).

The app is native internally in both modes; BlackBull(asgi=True) applies the native→ASGI boundary conversion in __call__ whenever the app is driven (scope entry included), so the app instance itself is the callable — uvicorn.run(app.to_asgi()) or uvicorn.run(app) are equivalent. Requires the asgi=True constructor flag: without it the app is wired for BlackBull's own native server.

`url_path_for(name, /, **params)` ¶

Return the path for the named route with params substituted.

`use(mw)` ¶

Register a global middleware applied to every non-lifespan request.

`warm_request(conn, *, body=b'', n=1)` `async` ¶

Invoke this app in-process n times with a :class:Connection to warm the request path.

A warm-up primitive: drives the full native __call__ → middleware → _dispatch chain (HTTP, gRPC, whatever conn routes to) with a synthetic receive that yields body once and a send that discards output. Faults in code pages and trips PEP 659 specialization on the dispatch + handler + codec — no socket, no wire I/O. Intended for use from an :meth:on_warmup hook; safe to call anytime.

Each iteration runs on a fresh copy of conn (its own body cache, state and receive binding), so the one template drives all n runs cleanly.

`CORS` ¶

Cross-Origin Resource Sharing (CORS) middleware.

Handles preflight OPTIONS requests and attaches CORS headers to actual cross-origin responses. Requests without an Origin header pass through unchanged.

Parameters:

Name	Type	Description	Default
`allow_origins`	`list[str] \| str`	Explicit origin strings or `['*']` for wildcard.	`'*'`
`allow_methods`	`list[str] \| None`	HTTP methods permitted in cross-origin requests. Defaults to `['GET', 'POST', 'HEAD', 'OPTIONS']`.	`None`
`allow_headers`	`list[str] \| str`	Request headers permitted; `['*']` allows all.	`'*'`
`allow_credentials`	`bool`	Emit `Access-Control-Allow-Credentials: true`. Cannot be combined with `allow_origins=['*']`.	`False`
`expose_headers`	`list[str] \| None`	Response headers the browser JS may read.	`None`
`max_age`	`int \| None`	Preflight cache lifetime in seconds. `None` omits the header.	`600`

Usage::

app = BlackBull()
app.use(CORS(
    allow_origins=['https://myapp.example.com'],
    allow_credentials=True,
    max_age=3600,
))

`ClientDisconnected` ¶

Bases: Exception

Raised when the client disconnects before the request body is complete.

ASGI signals a mid-body disconnect with an http.disconnect event that carries no body/more_body keys. Treating it as end-of-message would return a truncated upload as if it were whole, so :func:read_body raises this instead — the handler must not process a partial body as complete. The partial attribute holds whatever body bytes had arrived before the disconnect.

`Connection` `dataclass` ¶

One HTTP (or WebSocket) request — the single internal representation.

Built by the protocol actor, consumed by the router, dispatcher, middleware, and handlers. The ASGI scope dict is a derived view (:meth:as_scope). Request is a deprecated alias of this class.

`cookies` `property` ¶

Cookies from the Cookie header, parsed once and cached.

`disconnected` `property` ¶

True once the client dropped mid-request.

The named form of the state the module-level :func:disconnected helper also reports. A long-running handler polls this to abandon work whose answer nobody is waiting for::

for row in rows:
    if conn.disconnected:
        break

Set by the actor's disconnect-detecting receive wrapper, so it goes true when the server notices — at the next receive() — rather than the instant the peer's FIN lands.

The module-level :func:disconnected remains the form to use at the two ASGI boundaries, where the same state may live on a scope dict instead of a :class:Connection.

`path_params` `property` `writable` ¶

Matched URL path params, set by the router (values are converter- coerced, hence Any). The backing dict is created lazily on first access so no-param routes allocate nothing.

`query` `property` ¶

Query params as dict[str, str], parsed once and cached.

Last value wins on repeated keys (?tag=a&tag=b → {'tag': 'b'}). This is the honest exit for what the declared-parameter form cannot express — keys not known at registration time, and repeated keys; it does not coerce types (that is the declared form's job, which also answers a malformed value with 400 rather than a silent fallback).

`query_list` `property` ¶

Query params keeping every value, parsed once and cached.

The full parse_qsl result: ?tag=a&tag=b gives {'tag': ['a', 'b']}. Unlike :attr:query, which folds repeats to the last value, this is what list-valued keys need. The backing dict is created lazily on first access, so handlers that never read a query param allocate nothing.

`subprotocols` `property` ¶

The client-offered WebSocket subprotocols (ASGI websocket scope's subprotocols), parsed from the Sec-WebSocket-Protocol request header. Empty on HTTP requests and on a WS handshake that offered none. Derived — not stored — so it needs no ASGI round-trip and the header stays the single source of truth.

`as_scope()` ¶

Generate a fresh ASGI 3.0 scope dict (the single native→ASGI point).

Mutations of the returned dict do not affect the Connection, except that state and extensions are shared by reference so a buffering middleware's writes reach the handler.

`body()` `async` ¶

Return the complete request body, draining receive at most once.

Repeated calls (and :meth:json / :meth:text) return the cached bytes. A mid-body disconnect raises :class:ClientDisconnected. Raises :class:RuntimeError if :meth:stream already consumed the body — the channel is a single drain, so there is nothing left to buffer.

`form()` `async` ¶

application/x-www-form-urlencoded body, parsed once and cached.

Reads the body through :meth:body, so a later :meth:json / :meth:text call reuses the cached bytes. On a non-form Content-Type (or no body) it returns {} rather than raising, and does not touch the body. Multipart uploads are out of scope — use :meth:stream and parse them manually.

`from_scope(scope, receive=None)` `classmethod` ¶

Build a Connection from an external ASGI scope (the single ASGI→native point). Unknown keys are ignored; missing optional keys fall back to the field defaults.

`json()` `async` ¶

Parse the cached body as JSON (None on empty/invalid input).

`stream()` `async` ¶

Yield the request body one chunk at a time, draining receive once.

The streaming counterpart to :meth:body. Use it when the handler only needs to process the body incrementally — count/hash/forward a large upload — so the working set stays one chunk instead of the whole payload::

total = 0
async for chunk in conn.stream():
    total += len(chunk)

Mutually exclusive with :meth:body/:meth:json/:meth:text, which buffer: the body is a single-drain stream, so mixing the two on one request raises :class:RuntimeError rather than silently returning a partial or empty body. A mid-body disconnect raises :class:ClientDisconnected.

`text(encoding='utf-8')` `async` ¶

Decode the cached body as text (errors='replace').

`to_asgi_scope(*, force_asgi=False)` ¶

Materialize the ASGI scope the dispatch pipeline consumes, with this typed :class:Connection stashed on it for zero-reconversion reads.

The single canonical Connection → dispatch-ready scope bridge shared by the H/1.1 run() and H/2 _conn_to_scope seams, which would otherwise each hand-roll the same five steps:

derive the ASGI scope via :meth:as_scope (the one native→ASGI point);
when force_asgi (the §4.3 BB_FORCE_ASGI_SCOPE dual-path lane), round-trip through :meth:from_scope so both the derived scope and the Connection the consumers read are rebuilt from scratch on every request, keeping the compat conversion from bitrotting. _asterisk_form is Connection-only (not in the scope), so carry it across the rebuild;
restore scope['headers'] to the rich :class:Headers object (as_scope emits the ASGI list[tuple] form; internal .get() callers want the object);
re-expose the H/1.1 _asterisk_form OPTIONS marker on the envelope;
stash the Connection under :data:CONNECTION_STASH_KEY.

Protocol-specific augmentation (the websocket-only subprotocols key) is layered on by the caller after this returns — it is not a :class:Connection field (proposal §2.1).

The default (force_asgi=False) native path builds the scope by direct attribute access, placing the rich Headers object straight in (no list(headers) that the old code computed via the registry and then immediately discarded), and skips the per-field function-call indirection of as_scope(). The force_asgi dual-path conformance lane (§4.3) keeps the full as_scope → from_scope → as_scope round-trip so the compat conversion is still exercised.

`Depends` ¶

Declare a per-request provider for one simplified-handler parameter.

Use as the parameter's default value: db=Depends(get_db).

Provider forms (detected once, here):

async generator — yields the injected value exactly once; the code after yield (or the finally block) runs after the response has been sent, LIFO when several providers are active.
async function — awaited for the value; no cleanup.
sync function — called for the value; no cleanup.

Parameters:

Name	Type	Description	Default
`provider`	`Callable[[], Any]`	Zero-parameter callable in one of the three forms above.	required
`use_cache`	`bool`	When `True` (default), parameters of one handler that name the same provider share a single instance per request; `use_cache=False` calls the provider once per parameter.	`True`

Raises:

Type	Description
`TypeError`	At construction, when provider is not callable, takes parameters (including a nested `Depends` default — not supported in v1), or is a sync generator function.

`Event` `dataclass` ¶

An immutable message dispatched through EventDispatcher.

Attributes:

Name	Type	Description
`name`	`str`	The event name (e.g. `"app_startup"`).
`detail`	`dict`	Arbitrary per-event data. `detail` is used (rather than `payload`) to avoid colliding with HTTP/2 and WebSocket protocol terminology already used in the codebase.

`EventSourceResponse` ¶

Bases: StreamingResponse

Stream a Server-Sent Events response from an async iterator.

Yields are formatted per WHATWG §9.2.6 (the EventSource spec). Each item produced by content may be a str (bare data), bytes (bare data, UTF-8), or a Mapping with optional data / event / id / retry keys.

The content-type is forced to text/event-stream and Cache-Control: no-cache is auto-emitted; both are overridable via the headers argument if a deployment knows what it's doing.

Usage::

async def tokens():
    yield {'event': 'token', 'data': 'hello'}
    yield {'event': 'token', 'data': 'world'}
    yield {'event': 'done',  'data': ''}

@app.route(path='/sse')
async def stream():
    return EventSourceResponse(tokens())

`HTTPException` ¶

Bases: Exception

An exception carrying the HTTP status the dispatcher should report.

Raising this from a handler — or from the framework's own request-body adapter — makes BlackBull._dispatch answer with status instead of the generic 500, and (for 4xx) log it quietly as a client error rather than dumping a server traceback. detail is an optional human-readable message surfaced in development-mode error pages.

`Headers` ¶

Ordered multi-valued HTTP header store.

Satisfies the ASGI Iterable[[byte string, byte string]] contract while also providing O(1) dict-like lookup.

Invariants:

Header names and values are always bytes (per ASGI spec).
Lookups are case-insensitive: the internal index is keyed on name.lower() (RFC 7230 §3.2 — header field names are case-insensitive). __contains__, __getitem__, getlist, and get accept any casing; iteration preserves the original casing of the input.
Insertion order of duplicate names is preserved (RFC 7230 §3.2.2).

Examples::

headers = Headers([(b'set-cookie', b'a=1'), (b'set-cookie', b'b=2')])

list(headers)
# [(b'set-cookie', b'a=1'), (b'set-cookie', b'b=2')]   # ASGI iteration

headers.getlist(b'set-cookie')
# [(b'set-cookie', b'a=1'), (b'set-cookie', b'b=2')]

headers.getlist(b'missing')
# []

headers.get(b'host')          # first value, or default
# b'localhost:8000'

`add(other)` ¶

Return a new Headers containing all pairs from self then other.

`eq(other)` ¶

Value equality on the ordered (name, value) pair list.

Two Headers are equal when they carry the same fields in the same order (RFC 7230 §3.2.2 — order is significant for repeated fields). Enables Connection round-trip equality.

`getitem(name)` ¶

Return all pairs for name. Raises KeyError if absent.

`append(name_or_pairs, value=None)` ¶

Append header(s) to the end of the list.

Two-argument form: append(name, value) — adds a single pair. One-argument form: append(pairs) — adds every pair in the iterable.

`from_lowered(pairs)` `classmethod` ¶

Build from pairs whose names are already lowercase.

The caller must guarantee that; nothing here checks it, and a name containing uppercase would be indexed unreachably (every accessor lowercases before its fallback probe, so the field would be invisible to lookup while still appearing in iteration).

Two callers can guarantee it. http1_actor._parse lowercases each name while validating it, so re-lowercasing in __init__ recomputes a known answer. HTTP/2 field names are lowercase by protocol — RFC 9113 §8.2.1 makes an uppercase name malformed, and HeadersFrame.parse_payload rejects the frame before any pair reaches the header list.

Takes ownership of pairs rather than copying it; the parser builds a throwaway list per request, and the copy is the point of the shortcut. Do not pass a list you intend to keep mutating.

`get(name, default=b'')` ¶

Return the first value for name, or default if absent.

Mirrors dict.get(key, default): single value, optional default. For headers that may repeat use getlist(name).

`get_sf_dict(name)` ¶

Parse name as a Structured Field Dictionary (RFC 9651).

Multiple field lines are combined first. Returns an ordered dict of member name → Item / Inner List, or None if the field is absent or fails strict parsing (per RFC 9651 §4.2 the whole field is then ignored).

Example::

headers.get_sf_dict(b'priority')      # {'u': (2, {}), 'i': (True, {})}

`get_sf_item(name)` ¶

Parse name as a Structured Field Item (RFC 9651).

Returns (bare_item, parameters), or None if the field is absent or fails strict parsing (per RFC 9651 §4.2 the whole field is then ignored).

Example::

headers.get_sf_item(b'deprecation')   # (Date(1659578233), {})

`get_sf_list(name)` ¶

Parse name as a Structured Field List (RFC 9651).

Multiple field lines are combined first. Returns a list of Items / Inner Lists, or None if the field is absent or fails strict parsing (per RFC 9651 §4.2 the whole field is then ignored).

Example::

headers.get_sf_list(b'accept-query')  # [('a', {}), ('b', {})]

`getlist(name)` ¶

Return all pairs for name, or [] if the header is absent.

`InheritedFd` `dataclass` ¶

A socket already bound and listening, handed over by a supervisor.

Covers systemd-style activation and the re-exec handoff that keeps the listener continuous across an auto-reload.

`JSONResponse` ¶

Bases: Response

HTTP response with JSON-serialised body and application/json content-type.

Pass directly to the ASGI send callable when using BlackBull::

await send(JSONResponse({'ok': True}))
await send(JSONResponse({'error': 'Not found'}, status=HTTPStatus.NOT_FOUND))

`Listener` `dataclass` ¶

One listening socket and what happens on it.

speaks is always a positive name: 'http' selects the stack that detects HTTP/1.1, h2c and WebSocket upgrades, and a raw protocol names itself.

tls is the listener's own context, so a second certificate — or mTLS on one port and not another — is sayable.

workers is where ownership is decided, and the only place. Left unset it follows speaks: the HTTP stack is stateless and runs on every worker, while a raw protocol holds state that scattering across workers would scatter with it. Pass it explicitly to override.

`RedirectResponse` ¶

Bases: Response

HTTP redirect response carrying a Location header.

Completes the Response convenience family alongside JSONResponse. The body is empty; url becomes the Location header value and status a 3xx redirect code (default 302 Found — the safer general-purpose default, since it does not force the client to preserve the request method).

Pass directly to the ASGI send callable, or return it from a handler::

await send(RedirectResponse('/new-url'))
return RedirectResponse('/permanent', status=HTTPStatus.MOVED_PERMANENTLY)

url must be ASCII (RFC 9110 §10.2.2 — the Location field value is a URI-reference); percent-encode non-ASCII URLs before passing them in.

`Response` ¶

HTTP response object carrying body, status, and headers.

Pass directly to the ASGI send callable when using BlackBull::

await send(Response('<h1>Hello</h1>'))
await send(Response(b'data', status=HTTPStatus.NOT_FOUND))

`call(conn, receive, send)` `async` ¶

Drive this response as an ASGI app: emit start then body.

Mirrors :class:StreamingResponse so every BlackBull response type shares one protocol — await response(conn, receive, send) — whether returned from a simplified handler, invoked explicitly by a full-form handler, or normalised by app._wrap_send. Keeping the start/body serialisation here means there is a single source of truth for turning a Response into ASGI events.

`to_native()` ¶

Convert this response to the unified native message (one send).

The native-path serialiser: a complete Response becomes a single :class:~blackbull.native.NativeResponse carrying status, headers, and body — one object, one send. Symmetric with :meth:NativeResponse.to_asgi (the boundary conversion); streaming response types drive themselves and are not converted here.

`ResponseBody` ¶

Bases: dict

http.response.body event with typed property access.

`ResponseStart` ¶

Bases: dict

http.response.start event with typed property access.

`RouteInfo` ¶

Bases: NamedTuple

Immutable snapshot of a single registered route entry.

Returned by :meth:BlackBull.get_routes. One entry is produced per (route, method) pair — a route registered with methods=[GET, POST] yields two RouteInfo records.

Attributes:

Name	Type	Description
`method`	`str`	HTTP method string (e.g. `"GET"`, `"BREW"`).
`path`	`str`	URL template (e.g. `"/api/echo/{name}"`).
`name`	`str`	Endpoint name, or `""` if the route was registered unnamed.

`StreamingResponse` ¶

Stream a response body from an async generator.

Usage::

async def lines():
    for i in range(10):
        yield f'line {i}\n'.encode()
        await asyncio.sleep(0.1)

@app.route(path='/stream')
async def handler(conn, receive, send):
    await StreamingResponse(lines())(conn, receive, send)

`Tcp` `dataclass` ¶

A TCP port, on host or on every interface when it is None.

port=0 asks the OS for a free one; the bound port is read back after binding.

`TrustedProxy` ¶

Rewrite conn['client'] and conn['scheme'] from proxy headers.

Applied only when the direct TCP peer matches the configured trusted set, preventing malicious clients from spoofing X-Forwarded-For.

Supported headers (in precedence order):

RFC 7239 Forwarded — for=<ip>; proto=<scheme>
X-Forwarded-For — comma-separated IP chain; leftmost non-trusted IP wins
X-Forwarded-Proto — rewrite conn['scheme']

Parameters:

Name	Type	Description	Default
`trusted_proxies`	`list[str] \| str \| None`	IP addresses or CIDR strings (IPv4 or IPv6). Accepts a single string or a list. Defaults to loopback (`'127.0.0.1'`, `'::1'`).	`None`

Usage::

app = BlackBull(trusted_proxies=['127.0.0.1', '10.0.0.0/8'])

# or register explicitly for more control:
from blackbull import TrustedProxyMiddleware
app.use(TrustedProxyMiddleware(['127.0.0.1', '::1']))

`Unix` `dataclass` ¶

An AF_UNIX path.

`UnprocessableQuery` ¶

Bases: HTTPException

RFC 10008 §2.2 — a well-formed QUERY whose contents cannot be processed.

Raise this from a QUERY handler when the request media type was accepted (so 400/415 do not apply) but the query itself is semantically invalid — references an unknown field, violates a constraint, etc. The dispatcher answers 422 Unprocessable Content and, being an :class:HTTPException subclass, it flows through the normal error-router path and quiet 4xx logging.

`WebSocket` ¶

One WebSocket connection, as an object.

Constructed by the framework and handed to handlers that ask for it by annotation (ws: WebSocket) or by name (ws / websocket). A handler that takes the raw (conn, receive, send) triplet keeps receiving exactly that, and this object is never built for it.

The handshake is explicit: the connection is not live until :meth:accept returns, and sending before that is an error. To reject a connection, call :meth:close instead of :meth:accept.

`accepted` `property` ¶

True once :meth:accept has completed the handshake.

`client` `property` ¶

(host, port) of the peer; the port is None on a UDS.

`client_disconnected` `property` ¶

True once the peer's close has been observed.

Only ever set by reading — the disconnect arrives on the receive channel, so a handler that has stopped receiving will not see this flip. Use it to break out of a send-only loop that also receives.

`close_code` `property` ¶

The close code, once either side has closed; None while open.

`connection` `property` ¶

The underlying :class:~blackbull.connection.Connection.

Everything the handshake carried — headers, cookies, TLS state, the client address — is reachable through it. The shortcuts below cover the fields WebSocket handlers actually reach for.

`subprotocols` `property` ¶

Subprotocols the client offered, in its order of preference.

`aiter()` ¶

Iterate messages until the peer disconnects.

The disconnect ends the loop rather than raising, so the common shape is just async for message in ws:.

`accept(subprotocol=None, *, headers=None)` `async` ¶

Complete the handshake.

subprotocol names the one being accepted from :attr:subprotocols; leaving it None keeps the server's automatic negotiation, exactly as sending {'type': 'websocket.accept', 'subprotocol': None} does on the raw path. headers are extra response headers for the 101.

Raises :class:WebSocketDisconnect if the peer abandoned the handshake before it could be completed.

When middleware already accepted for you (blackbull.middleware.websocket), a bare accept() is a no-op so the same handler body works with or without it. Asking for a specific subprotocol or extra headers in that situation raises instead of silently dropping the request — the 101 has already gone out and cannot carry them.

`close(code=_NORMAL_CLOSURE, reason=None)` `async` ¶

Close the connection, or reject the handshake.

Called before :meth:accept, this rejects the connection — the client's connect() fails rather than opening and immediately closing. Idempotent, and a no-op once the peer has already gone, so a finally: await ws.close() is always safe.

`receive()` `async` ¶

Await one complete message — str for text, bytes for binary.

Fragmented messages have already been reassembled, so what comes back is always a whole application message.

Raises :class:WebSocketDisconnect when the peer closes. Prefer async for unless you need the close code.

`receive_bytes()` `async` ¶

Await one message, requiring it to be binary.

`receive_json()` `async` ¶

Await one message and parse it as JSON (text or binary).

`receive_text()` `async` ¶

Await one message, requiring it to be text.

`send(data)` `async` ¶

Send data as text or binary, chosen by its Python type.

`send_bytes(data)` `async` ¶

Send one complete binary message.

`send_json(data, *, binary=False)` `async` ¶

JSON-serialise data and send it as one message.

Text by default, which is what browsers and most clients expect; pass binary=True to send the UTF-8 encoding as a binary frame instead.

`send_text(data)` `async` ¶

Send one complete text message.

`WebSocketDisconnect` ¶

Bases: Exception

The peer closed the connection.

Raised by :meth:WebSocket.receive and its typed variants. Iterating with async for handles this for you — the loop simply ends — so catch it only when you call receive() directly and need the close code.

code is the RFC 6455 close code the peer sent, or 1005 when it sent none. reason is its optional UTF-8 explanation.

`WebSocketResponse(content)` ¶

Build an ASGI websocket.send event dict from content.

str → {'type': 'websocket.send', 'text': content}
bytes → {'type': 'websocket.send', 'bytes': content}
anything else → JSON-serialised into the text field

Pass the result directly to the ASGI send callable::

await send(WebSocketResponse('hello'))

`getattr(name)` ¶

Lazy, deprecated attribute access — blackbull.Request.

Request was the opt-in HTTP context object; it has been merged into :class:Connection and the name demoted to an alias. Resolving it through the module __getattr__ (PEP 562) means the DeprecationWarning fires only if code actually touches Request — importing the package stays warning-free — and the alias still evaluates to Connection so existing request: Request handler signatures keep working unchanged during the migration window (removal no earlier than 2027-08-01).

`as_middleware(target)` ¶

Decorator that marks an async function or class as BlackBull middleware.

Wraps call_next so any send callable the middleware passes to it is automatically normalised — Response/JSONResponse objects are converted to NativeResponse before reaching the middleware's inner send wrapper. The wrapper therefore only ever sees the native representation (NativeResponse on the HTTP path — H1 and H2).

Applied to an async function (signature (conn, receive, send, call_next))::

@as_middleware
async def timing_mw(conn, receive, send, call_next):
    async def timed_send(event):
        # event is a NativeResponse on the HTTP path
        await send(event)
    await call_next(conn, receive, timed_send)

Applied to a class whose __call__ is the middleware coroutine::

@as_middleware
class Cache:
    async def __call__(self, conn, receive, send, call_next):
        async def cap_send(event):
            # event is a NativeResponse on the HTTP path
            ...
        await call_next(conn, receive, cap_send)

Power users who need to handle raw send arguments (e.g. because their middleware is used in a context where no simplified handlers are registered) should omit this decorator — their call_next is then wired directly to the next handler with no extra wrapping.

`cookie_header(name, value, path='/', http_only=True)` ¶

Build a set-cookie header tuple suitable for inclusion in response headers.

`cookies_from_headers(headers)` ¶

Parse the Cookie header(s) into a dict, straight from a headers object/iterable — no ASGI scope dict involved.

This is the native core: :meth:Connection.cookies calls it directly on conn.headers; :func:parse_cookies is the ASGI-scope-shaped wrapper kept for external callers that hold a scope dict. Accepts a :class:blackbull.headers.Headers instance (uses getlist) or a plain iterable of (name, value) bytes tuples (the ASGI 3.0 form).

`parse_cookies(source)` ¶

Parse the Cookie request header into a dict.

source is a mapping carrying a 'headers' key — an ASGI scope dict, or the {'headers': conn.headers} wrapper Connection.cookies passes. Works identically for HTTP/1.1, HTTP/2, and WebSocket. HTTP/1.1 sends a single combined Cookie header; HTTP/2 may split it into multiple fields (RFC 7540 §8.1.2.5). All cookie fields are collected and joined before parsing, so both wire formats produce the same result.

Accepts either of the two header shapes that may appear on source['headers']:

A plain list/iterable of (name, value) bytes tuples — the standard ASGI 3.0 form, used by external servers (uvicorn, hypercorn, httpx.ASGITransport).
A :class:blackbull.headers.Headers instance — what BlackBull's own server attaches as a handler ergonomics enhancement.

`parse_response_event(event)` ¶

Wrap event in the appropriate typed subclass for dispatch.

The returned object IS the event dict (shallow copy) — pass it directly to downstream send callables without re-serialisation. Trailers and unknown event types are returned unchanged.

ResponseStart/ResponseBody are dict subclasses, not statically members of ASGISendEvent; a caller threading the result back into a typed send callable casts at that seam (a runtime no-op) rather than widening the public union.

`read_body(receive)` `async` ¶

Read the complete request body from the ASGI receive channel.

Collects chunks in a list and joins once, rather than the O(n²) += growth. A single-chunk body (the common case) is returned directly with no intermediate copy at all.

Raises :class:ClientDisconnected if an http.disconnect arrives before the body is complete, so a truncated upload is never silently returned as if whole.

`read_json(receive)` `async` ¶

Read the request body and parse it as JSON.

Returns the parsed JSON value (dict, list, str, int, float, bool), or None when the body is empty, not valid JSON, or not decodable. Callers should treat None as a client error and respond 400::

data = await read_json(receive)
if data is None:
    await send(JSONResponse({'error': 'invalid JSON'},
                            status=HTTPStatus.BAD_REQUEST))
    return

Note that a literal JSON null body also parses to None; if that distinction matters, read the body yourself with :func:read_body.

A mid-body client disconnect propagates as :class:ClientDisconnected rather than being reported as invalid JSON — a truncated body is a transport failure, not a parse error.

`read_text(receive, encoding='utf-8')` `async` ¶

Read the request body and decode it as text.

Uses errors='replace' so undecodable bytes become U+FFFD rather than raising — a malformed body never crashes the handler. Override encoding for non-UTF-8 payloads.

A mid-body client disconnect still propagates as :class:ClientDisconnected: a truncated upload must not be decoded and returned as if it were the complete text.

`serve(app, *, certfile=None, keyfile=None, port=0, unix_path=None, inherited_fd=None, listeners=None, workers=None, max_connections=None, stream_queue_depth=None, ws_queue_depth=None, reload=False, reload_paths=None)` ¶

Synchronous entry point for any ASGI 3.0 callable.

Works for a :class:BlackBull instance and for any plain ASGI callable (uvicorn/hypercorn-style async def app(scope, receive, send): …). This is what the blackbull console script calls after resolving module:attr; :meth:BlackBull.serve is a thin shim around it.

For workers=1 without reload the server runs in the current process via asyncio.run. For workers > 1 or reload=True the master pre-binds sockets, forks workers, and blocks until SIGTERM / SIGINT — or in reload mode, until a watched file changes (master then re-execs itself, see :mod:blackbull.server.reload).

listeners states the sockets directly — one :class:~blackbull.server.listener.Listener each, saying where it is, what speaks there and whether TLS terminates there. It replaces port, unix_path and inherited_fd rather than joining them, so passing both is refused.

All integer parameters default to their corresponding BB_* environment variables (see :mod:blackbull.env).

blackbull¶

blackbull ¶

QUERY = 'QUERY' module-attribute ¶

AppConfig dataclass ¶

BlackBull ¶

add_extension(ext) ¶

drain_events(timeout=5.0) async ¶

enable_grpc(registry) ¶

enable_openapi(*, title='BlackBull API', version='0.1.0', description=None, spec_path='/openapi.json', docs_path='/docs') ¶

get_routes() ¶

group(middlewares=[]) ¶

intercept(event_name) ¶

on(event_name, *, blocking=False) ¶

on_error(key) ¶

on_shutdown(fn) ¶

on_startup(fn) ¶

on_warmup(fn) ¶

raw_handler(name, *, port=None, detector=None, tls=False, stateful=True) ¶

register_converter(type_, converter=None) ¶

register_protocol_handler(name, handler, *, detector=None, port=None, tls=False, stateful=True) ¶

route(methods=[HTTPMethod.GET], path='/', scheme=Scheme.http, functions=[], middlewares=[], name=None, accept_query=None) ¶

run(certfile=None, keyfile=None, port=None, unix_path=None, inherited_fd=None, listeners=None, workers=None, max_connections=None, stream_queue_depth=None, ws_queue_depth=None, reload=None, reload_paths=None) ¶

static(url_prefix, root_dir, *, cache=False, index=None, conditional=True) ¶

to_asgi() ¶

url_path_for(name, /, **params) ¶

use(mw) ¶

warm_request(conn, *, body=b'', n=1) async ¶

CORS ¶

ClientDisconnected ¶

Connection dataclass ¶

cookies property ¶

disconnected property ¶

path_params property writable ¶

query property ¶

query_list property ¶

subprotocols property ¶

as_scope() ¶

body() async ¶

form() async ¶

from_scope(scope, receive=None) classmethod ¶

json() async ¶

stream() async ¶

text(encoding='utf-8') async ¶

to_asgi_scope(*, force_asgi=False) ¶

Depends ¶

Event dataclass ¶

EventSourceResponse ¶

HTTPException ¶

Headers ¶

__add__(other) ¶

__eq__(other) ¶

__getitem__(name) ¶

append(name_or_pairs, value=None) ¶

from_lowered(pairs) classmethod ¶

get(name, default=b'') ¶

get_sf_dict(name) ¶

get_sf_item(name) ¶

get_sf_list(name) ¶

getlist(name) ¶

InheritedFd dataclass ¶

JSONResponse ¶

Listener dataclass ¶

RedirectResponse ¶

Response ¶

__call__(conn, receive, send) async ¶

to_native() ¶

ResponseBody ¶

ResponseStart ¶

RouteInfo ¶

StreamingResponse ¶

Tcp dataclass ¶

TrustedProxy ¶

Unix dataclass ¶

UnprocessableQuery ¶

WebSocket ¶

accepted property ¶

client property ¶

client_disconnected property ¶

close_code property ¶

connection property ¶

`blackbull`¶

`blackbull` ¶

`QUERY = 'QUERY'` `module-attribute` ¶

`AppConfig` `dataclass` ¶

`BlackBull` ¶

`add_extension(ext)` ¶

`drain_events(timeout=5.0)` `async` ¶

`enable_grpc(registry)` ¶

`enable_openapi(*, title='BlackBull API', version='0.1.0', description=None, spec_path='/openapi.json', docs_path='/docs')` ¶

`get_routes()` ¶

`group(middlewares=[])` ¶

`intercept(event_name)` ¶

`on(event_name, *, blocking=False)` ¶

`on_error(key)` ¶

`on_shutdown(fn)` ¶

`on_startup(fn)` ¶

`on_warmup(fn)` ¶

`raw_handler(name, *, port=None, detector=None, tls=False, stateful=True)` ¶

`register_converter(type_, converter=None)` ¶

`register_protocol_handler(name, handler, *, detector=None, port=None, tls=False, stateful=True)` ¶

`route(methods=[HTTPMethod.GET], path='/', scheme=Scheme.http, functions=[], middlewares=[], name=None, accept_query=None)` ¶

`run(certfile=None, keyfile=None, port=None, unix_path=None, inherited_fd=None, listeners=None, workers=None, max_connections=None, stream_queue_depth=None, ws_queue_depth=None, reload=None, reload_paths=None)` ¶

`static(url_prefix, root_dir, *, cache=False, index=None, conditional=True)` ¶

`to_asgi()` ¶

`url_path_for(name, /, **params)` ¶

`use(mw)` ¶

`warm_request(conn, *, body=b'', n=1)` `async` ¶

`CORS` ¶

`ClientDisconnected` ¶

`Connection` `dataclass` ¶

`cookies` `property` ¶

`disconnected` `property` ¶

`path_params` `property` `writable` ¶

`query` `property` ¶

`query_list` `property` ¶

`subprotocols` `property` ¶

`as_scope()` ¶

`body()` `async` ¶

`form()` `async` ¶

`from_scope(scope, receive=None)` `classmethod` ¶

`json()` `async` ¶

`stream()` `async` ¶

`text(encoding='utf-8')` `async` ¶

`to_asgi_scope(*, force_asgi=False)` ¶

`Depends` ¶

`Event` `dataclass` ¶

`EventSourceResponse` ¶

`HTTPException` ¶

`Headers` ¶

`add(other)` ¶

`eq(other)` ¶

`getitem(name)` ¶

`append(name_or_pairs, value=None)` ¶

`from_lowered(pairs)` `classmethod` ¶

`get(name, default=b'')` ¶

`get_sf_dict(name)` ¶

`get_sf_item(name)` ¶

`get_sf_list(name)` ¶

`getlist(name)` ¶

`InheritedFd` `dataclass` ¶

`JSONResponse` ¶

`Listener` `dataclass` ¶

`RedirectResponse` ¶

`Response` ¶

`call(conn, receive, send)` `async` ¶

`to_native()` ¶

`ResponseBody` ¶

`ResponseStart` ¶

`RouteInfo` ¶

`StreamingResponse` ¶

`Tcp` `dataclass` ¶

`TrustedProxy` ¶

`Unix` `dataclass` ¶

`UnprocessableQuery` ¶

`WebSocket` ¶

`accepted` `property` ¶

`client` `property` ¶

`client_disconnected` `property` ¶

`close_code` `property` ¶

`connection` `property` ¶