blackbull.app

blackbull.app

BlackBull application object — the user-facing ASGI 3.0 entry point.

RouteGroup, the default error handler and the handler-boundary send adapter live here rather than in the modules they belong to: importing them from there would close an import cycle back to this one.

BlackBull

The application: a router, its error handlers, and the ASGI callable.

Register routes with route, or with group when several should share a middleware prefix; global middleware with use; lifecycle hooks with on_startup, on_shutdown and on_warmup. run serves the app on BlackBull's own server, and the instance is itself a plain ASGI 3.0 callable, so an external host can drive it instead.

Protocols other than HTTP and WebSocket attach through add_extension or register_protocol_handler; nothing here is added by subclassing.

Attributes:

Name	Type	Description
extensions	dict[str, object]	Installed extensions by name. The Extensions guide says what an extension puts there and how to reach it.

available_ws_protocols property writable

WebSocket subprotocols this app offers, as bytes.

Assigning a list of str encodes each entry.

loop property

The bound event loop, or None when none is running yet.

Read from synchronous setup code this answers None rather than raising, so a caller that does not need a loop still works; asyncio supplies one once the coroutines run.

__call__(conn, receive, send) async

The ASGI 3.0 callable, and the app's only native/ASGI boundary.

Everything below this method threads a Connection, in both modes. An ASGI scope dict arrives only from an external host (uvicorn, httpx.ASGITransport) or under BB_FORCE_ASGI_SCOPE=1, and is converted here, once; the same edge wraps the host's send with the native→ASGI conversion, while BlackBull's own server passes a native-capable send through untouched. Lifespan is the one scope that is never a request and never becomes a Connection.

Parameters:

Name	Type	Description	Default
conn	Connection | dict	A Connection, or an ASGI scope dict from a host.	required
receive	ASGIReceiveCallable | None	Optional because a handler that never reads a body never touches it — the router guards on a missing receive and dispatch completes normally. A conforming ASGI host always passes a callable; the tolerance is for direct drives.	required
send	ASGISendCallable	The host's send callable.	required

__init__(loop=None, observer_shutdown_timeout=5.0, trusted_proxies=None, config=None, cache_max=None, asgi=False)

Build an application.

Parameters:

Name	Type	Description	Default
loop	AbstractEventLoop | None	Event loop to bind to. Left unset, loop picks up the running loop once there is one.	None
observer_shutdown_timeout	float	Seconds to wait at shutdown for detached @app.on observers to finish.	5.0
trusted_proxies	list[str] | str | None	Addresses or CIDRs whose forwarded headers are honoured; installs TrustedProxy.	None
config	AppConfig | None	Deploy settings run falls back to. The Configuration guide gives the resolution order.	None
cache_max	int | None	Route-resolution cache size. None keeps the router's own default.	None
asgi	bool	Mark the app as driven by an external ASGI host, so __call__ converts at the boundary in both directions.	False

add_extension(ext)

Call ext.init_app(app) immediately and return ext.

Subclassing Extension is optional. Optional async startup(app)/shutdown(app) methods are registered as lifespan hooks.

drain_events(timeout=5.0) async

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

Only detached observers need waiting for; the events guide says which do not. False means timeout expired with work still outstanding. Nothing is cancelled; call again with a longer budget.

enable_grpc(registry)

Serve all four gRPC call shapes from registry over HTTP/2.

Handlers exchange raw message bytes without a protobuf dependency. Use TLS+ALPN or h2c; status is sent in trailers. HTTP routes remain available.

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

Publish OpenAPI JSON at spec_path and Swagger UI at docs_path.

Call once. The spec is regenerated per request to include new routes. docs_path=None disables the UI; see docs/guide/openapi.md.

get_routes()

Return a shallow snapshot in registration order, one RouteInfo per HTTP method.

Changing the returned list cannot mutate the live router.

group(middlewares=[])

Return a RouteGroup that prepends middlewares to every route.

intercept(event_name)

Await interceptors in registration order; return the handler unchanged.

Exceptions propagate and abort subsequent interceptors for that event.

on(event_name, *, blocking=False)

Observe an event; the decorated coroutine is returned unchanged.

The default schedules a detached task. blocking=True awaits observers in registration order; cancellation can still interrupt them. Ordinary observer exceptions are logged and isolated. Use intercept for hooks that may change the request, and see docs/guide/events.md for event lifetimes.

on_error(key)

Register a (conn, receive, send) handler for a status or exception class.

An int key is converted to HTTPStatus. conn.state supplies error_status, error_exception when an exception triggered it, and allowed_methods for 405. See docs/guide/error-handling.md for exception-class precedence.

on_shutdown(fn)

Register a zero-argument coroutine; return it unchanged.

Run in registration order before shutdown.complete. An exception aborts remaining hooks.

on_startup(fn)

Register a zero-argument coroutine; return it unchanged.

Run in registration order before startup.complete. An exception aborts remaining hooks and withholds the acknowledgement.

on_warmup(fn)

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

The hook runs once in the master, before the listening socket is created and before workers fork; on_startup runs per worker, after both. 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 on_startup. Use warm_request to exercise the ASGI dispatch/handler path in-process, and 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. The Workers deployment page shows a hook and explains what makes a warmed master's heap survive the fork.

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

Decorator form of 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)

Convert a custom simplified-handler return type to a supported sendable.

The converter must return Response, str, bytes, None, or a JSON-able mapping, list or dataclass. Pass it directly or omit it to get a decorator. Converters registered after a route still apply to that route.

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 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 synchronously; do not wrap in asyncio.run.

Unset options resolve through BLACKBULL_* environment variables, .env, AppConfig and defaults. BB_* controls server tuning. Multiple workers or reload use a blocking supervisor. External ASGI hosts can call the app; embedded event-loop use requires ASGIServer. See the Configuration guide.

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

Mount static GET/HEAD routes under url_prefix.

cache=True retains file bodies in memory. index names a directory index; None disables it. conditional=False disables ETag/Last-Modified and 304 responses. See docs/guide/static-files.md for cache and traversal limits.

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 the native app n times without sockets, discarding output.

Each iteration copies conn with independent body/state/receive bindings. The synthetic receive yields body once. Intended for on_warmup hooks.

RouteGroup

A subset of routes that share a common middleware prefix.

Obtain via app.group(middlewares=[...]). Every route registered through the group automatically prepends the group middlewares before any per-route middlewares.

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

Register a route on this group, prepending the group middlewares.

Same parameters as BlackBull.route. accept_query is the RFC 10008 list of request media types the route accepts (the Accept-Query response header + Content-Type enforcement) — it is a content-negotiation policy, not a switch for the QUERY method (a method is accepted purely by listing it in methods).

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)

Run a BlackBull instance or ASGI 3.0 callable synchronously.

One worker without reload runs in this process; multiple workers or reload use a blocking supervisor. listeners replaces port/unix_path/inherited_fd; passing both is refused. Unset workers, max_connections and queue depths use BB_* settings. Deployment options do not resolve BLACKBULL_* here.