`blackbull.app`¶

`blackbull.app` ¶

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

Exposes the BlackBull class which wraps a Router, an ErrorRouter, lifespan hooks, per-route and global middleware chains, and (via app.static) static-file serving. BlackBull.__call__ is the ASGI callable: it dispatches lifespan events to _handle_lifespan and routes HTTP / WebSocket scopes through the global-middleware chain ending in _dispatch.

Companion definitions live in this module to avoid circular imports:

RouteGroup — returned by app.group(middlewares=[...]) to share a middleware prefix across routes.
_default_error_handler — registered on every HTTPStatus error and on Exception so unhandled errors produce a sensible plain-text reply.
_wrap_send — adapts the ASGI send callable so handlers may pass Response objects directly.

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

`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)` ¶

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)` ¶

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

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`	Reserved for first-byte sniffing on shared ports (Sprint 51); unused today.	`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 (Sprint 75).	`False`

`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, 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 via global middleware.

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.

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

`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 :meth: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, 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).

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

blackbull.app¶

blackbull.app ¶

BlackBull ¶

add_extension(ext) ¶

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) ¶

register_converter(type_, converter=None) ¶

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

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, 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) ¶

url_path_for(name, /, **params) ¶

use(mw) ¶

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

RouteGroup ¶

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

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

`blackbull.app`¶

`blackbull.app` ¶

`BlackBull` ¶

`add_extension(ext)` ¶

`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)` ¶

`register_converter(type_, converter=None)` ¶

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

`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, 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)` ¶

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

`use(mw)` ¶

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

`RouteGroup` ¶

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

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