Requests and responses¶

How to read what the client sent, build the response, stream a body back, and detect when the client has gone away.

The `Connection` object¶

The easiest way to read the request is the opt-in Connection context object — BlackBull's single internal request representation (the ASGI scope is a derived view of it). Declare a parameter annotated Connection (any name) — or a parameter named request/conn with no annotation — and the router injects one, the same way it injects path params and body:

from blackbull import BlackBull, Connection

app = BlackBull()

@app.route(path='/users/{uid}', methods=[HTTPMethod.POST])
async def show(uid: int, conn: Connection):
    token = conn.headers.get(b'authorization')   # Headers view
    who   = conn.client                          # (host, port) or None
    lang  = conn.cookies.get('lang', 'en')       # dict[str, str]
    data  = await conn.json()                    # parsed once, cached
    return {'uid': uid, 'lang': lang, 'data': data}

Read-side surface:

member	value
`conn.method` / `conn.path` / `conn.scheme`	`str` (`path` is percent-decoded — see below)
`conn.client`	`(host, port)` tuple, or `None`
`conn.path_params`	`dict[str, str]` of matched `{placeholder}` values
`conn.headers`	case-insensitive `Headers` view
`conn.cookies`	`dict[str, str]`, parsed once (all protocols)
`await conn.body()`	complete body as `bytes`, buffered once and cached
`await conn.json()`	parsed JSON, or `None` on empty/invalid body
`await conn.text(encoding='utf-8')`	body decoded as text (`errors='replace'`)
`conn.as_scope()`	a freshly-derived ASGI scope dict (escape hatch)

body() drains the receive channel at most once; json() and text() share the same cache, and a handler that also declares body: bytes (or a dataclass body parameter) receives the same cached bytes — the body is never read twice. Handlers that don't declare a Connection pay nothing: injection is decided when the route is registered, and the raw (scope, receive, send) form is unaffected.

Request is deprecated — use Connection (since v0.60.0)

Connection replaces the old Request context object. blackbull.Request is now a deprecated alias of Connection that emits a DeprecationWarning on first use and will be removed no earlier than 2027-08-01. Migrate by renaming the import and the annotation — the members are identical:

# before                          # after
from blackbull import Request      from blackbull import Connection

async def h(request: Request):     async def h(conn: Connection):
    ...                                ...

One member changed name: the raw-scope escape hatch was request.scope (a stored dict) and is now conn.as_scope() (a freshly-derived dict) — the scope is no longer the primary representation, so it is generated on demand rather than held.

path is percent-decoded (since v0.53.0)

conn.path (and the underlying scope['path']) is the percent-decoded request target with the query string removed — /files/a%2Fb arrives as /files/a/b, matching the ASGI spec and uvicorn. The undecoded bytes are available as conn.raw_path (bytes, query excluded) for the rare consumer — reverse proxies, WAFs, cache-key / signature verification — that must reproduce the exact received byte sequence. RFC 3986 ; parameters are preserved in both (/cart;sid=abc stays intact). Before v0.53.0 path was not decoded; code that re-decoded it itself should drop that step.

The sections below cover the same reads on the raw ASGI surface — useful inside middleware (which always uses the full form) and for streaming bodies chunk by chunk.

Reading the request body¶

from blackbull import read_body

@app.route(path='/echo', methods=[HTTPMethod.POST])
async def echo(conn, receive, send):
    raw: bytes = await read_body(receive)
    await send(Response(raw))

read_body reads all body chunks until more_body=False and returns a single bytes object. The stream is consumed — call at most once per request (typically inside a middleware, not the handler itself).

For streaming uploads, call receive() directly — each call returns one chunk with more_body=True until the final chunk arrives with more_body=False. read_body is a convenience wrapper that buffers all chunks before returning.

The body has a ceiling, and it is not the handler's job¶

Two limits apply before your handler sees a byte, on HTTP/1.1 and HTTP/2 alike:

BB_MAX_BODY_SIZE (30 MiB) — the total the request may deliver. A declared Content-Length over the cap is answered 413 Content Too Large without reading the body at all, so the handler is never entered; an undeclared or chunked body over it is refused mid-stream.
BB_MIN_BODY_RATE (240 B/s, after a 5 s grace) — the slowest a peer may deliver. Below it the request is abandoned, the same way BB_BODY_TIMEOUT abandons a silent one.

An upload endpoint that legitimately accepts more raises the cap (BB_MAX_BODY_SIZE=104857600); 0 disables it and hands the 413 decision back to the application. Both are documented in full under environment variables.

A handler can still see a 413: when a chunked HTTP/1.1 body crosses the cap mid-stream, receive() raises HTTPException(413), which the dispatcher answers like any other status-carrying error — so an @app.on_error(413) handler is honoured.

Noticing that the client left¶

conn.disconnected is True once the client dropped mid-request. It is worth checking in a loop whose result nobody is waiting for any more:

@app.route('/report')
async def report(conn):
    rows = []
    async for row in slow_query():
        if conn.disconnected:
            return            # nobody is listening; stop paying for this
        rows.append(row)
    return {'rows': rows}

The flag goes true when the server notices — at the next receive() — not the instant the peer's FIN arrives, so a handler that never reads the body and never streams will not see it change. For a push-style notification instead of a poll, use the request_disconnected event.

`read_json` and `read_text`¶

For the two most common body shapes, read_json and read_text wrap read_body so the handler skips the parse/decode boilerplate:

from http import HTTPStatus
from blackbull import read_json, read_text, JSONResponse

@app.route(path='/api/things', methods=[HTTPMethod.POST])
async def create_thing(conn, receive, send):
    data = await read_json(receive)        # dict | list | … | None
    if data is None:
        await send(JSONResponse({'error': 'invalid JSON'},
                                status=HTTPStatus.BAD_REQUEST))
        return
    await send(JSONResponse({'created': data}))

@app.route(path='/api/note', methods=[HTTPMethod.POST])
async def note(conn, receive, send):
    text = await read_text(receive)        # str (errors='replace')
    await send(Response(text))

read_json returns None when the body is empty, not valid JSON, or not decodable — treat None as a client error. read_text never raises on malformed bytes (undecodable bytes become U+FFFD); pass encoding= for non-UTF-8 payloads. Both consume the stream, so call at most once per request, exactly like read_body.

Recommended JSON-body middleware¶

import json
from http import HTTPStatus
from blackbull import read_body, JSONResponse

async def json_body_mw(conn, receive, send, call_next):
    raw = await read_body(receive)
    try:
        conn.state['json'] = json.loads(raw)
    except (json.JSONDecodeError, ValueError):
        await send(JSONResponse({'error': 'Invalid JSON'},
                                status=HTTPStatus.BAD_REQUEST))
        return
    await call_next(conn, receive, send)

The handler then reads conn.state['json'] without touching receive.

Reading request headers¶

conn.headers is a Headers object — case-insensitive, ordered, multi-valued.

# First value for a header; returns b'' when absent
ct   = conn.headers.get(b'content-type')
auth = conn.headers.get(b'authorization', b'')

# All (name, value) pairs for a header (multi-value support)
pairs = conn.headers.getlist(b'accept')      # list[tuple[bytes, bytes]]

# ASGI-compliant iteration
for name, value in conn.headers:
    ...

# Membership test
if b'content-length' in conn.headers:
    ...

Header names are stored lowercase.

`get` vs `getlist`¶

Use .get(name) for headers that appear at most once (content-type, authorization, host). Use .getlist(name) for headers that may repeat:

Header	Why it can repeat
`accept`, `accept-encoding`	Clients may send multiple preference lines
`set-cookie`	Servers send one `Set-Cookie` field per cookie
`cookie` (HTTP/2)	RFC 7540 §8.1.2.5 requires one field per cookie pair

.getlist returns list[tuple[bytes, bytes]] — the full (name, value) pairs in insertion order, or [] if the header is absent.

Why cookie needs getlist on HTTP/2

HTTP/1.1 combines all cookies into a single Cookie: a=1; b=2 field. HTTP/2 sends each cookie as a separate header field to enable HPACK compression of individual values. Calling .get(b'cookie') on an HTTP/2 scope silently discards all but the first cookie. parse_cookies (below) handles this correctly for all protocols.

Reading cookies¶

from blackbull import parse_cookies

cookies: dict[str, str] = parse_cookies(scope)
session = cookies.get('session', '')

parse_cookies(scope) returns a dict[str, str] of cookie name → value from the current request. The result is identical across HTTP/1.1, HTTP/2, and WebSocket scopes — you don't need to know how the client delivered the cookies.

Query parameters¶

Declare them as handler parameters (since v0.56.0). Any simplified-handler parameter that is not a path param, body, conn, Connection, a dataclass body, or Depends resolves from the query string, coerced to its annotation:

@app.route(path='/search')
async def search(q: str, page: int = 1, exact: bool = False):
    ...   # /search?q=bull&page=2 → q='bull', page=2, exact=False

Types: str (default for unannotated params), int, float, and bool (1/true/yes/on and 0/false/no/off, case-insensitive). T | None of a supported scalar also works. Anything else — containers, models — is a registration-time TypeError; drop to Connection or the raw scope for those.
Required vs optional: a parameter with a default is optional; without one, a request missing the key is answered with 400. A value that fails coercion (?page=abc for page: int) is also a 400 — client errors never surface as 500s.
Repeated keys (?tag=a&tag=b): the last occurrence wins. For list-valued keys, parse the raw query string as shown below.
Precedence: a parameter that matches a {placeholder} in the path is always a path param — the path value shadows any same-named query key (declaring a default on a path param draws a registration-time warning, since that usually means a query param was intended).
OpenAPI: query params appear in the generated spec (in: query, schema from the annotation, required from default-presence) — see OpenAPI.

Coercion and required-ness are resolved when the route is registered, not per request — handlers that declare no query params keep the exact adapted form they had before.

Parsed query params: `conn.query` / `conn.query_list`¶

When the declared form cannot express the request — repeated keys, keys not known at registration time — Connection parses the query string for you, lazily and once:

q = conn.query          # dict[str, str]; last value wins on ?tag=a&tag=b
ql = conn.query_list    # dict[str, list[str]]; keeps every value

conn.query folds repeated keys to the last value: /search?q=bull&tag=a&tag=b → {'q': 'bull', 'tag': 'b'}. Values are decoded, blank values are kept, and the result is cached on first access.
conn.query_list keeps every value: /search?tag=a&tag=b → {'tag': ['a', 'b']}. It is the escape hatch for list-valued keys.
Both are lazy — a handler that never reads a query param allocates nothing — and both are parsed once and cached, so re-reading them never re-parses the raw conn.query_string.

Neither coerces types. That is the declared form's job (q: int = 1), which also answers a malformed value with 400 rather than a silent fallback — so the division is: declared parameters are for typed scalars; conn.query is the honest exit for what the declared form cannot express. If you want a typed scalar, declare it; don't plumb conn.query for one.

For full control — unusual (non-UTF-8) encodings, or order-preserving pairs — conn.query_string (raw bytes) is still there; parse it with urllib.parse.parse_qsl.

Form data¶

HTML forms with enctype="application/x-www-form-urlencoded" (the default) send key=value pairs in the body. conn.form() parses and caches them:

form = await conn.form()   # dict[str, str]; last value wins on repeats
name = form.get('name', '')

Reads the body through conn.body(), so a later conn.json() / conn.text() call reuses the cached bytes.
On a non-form Content-Type (or no body) it returns {} rather than raising, and does not consume the body.
Like conn.query, it is parsed once and cached.

Multipart file uploads (multipart/form-data) are not yet supported by a built-in helper. Use the python-multipart package to parse the body manually.

Responses¶

Native send path

On BlackBull's own HTTP server (HTTP/1.1 and HTTP/2), send accepts Response / JSONResponse objects, (bytes, status, headers), NativeResponse, and ASGI http.response.* dicts — the full-form ASGI dict forms are a compatibility contract held until 2027-07-29, converted to the native message at the handler boundary. A complete Response becomes one NativeResponse object (one send); streaming is a header object then body-chunk objects. Under an external ASGI host (BlackBull(asgi=True) + to_asgi()) the same handler code runs with plain ASGI dicts on the wire. The http.response.start trailers flag is preserved losslessly via NativeResponse.expects_trailers, and http.response.trailers more_trailers is preserved via NativeResponse.more_trailers.

HTTP/1.1 framing is owned by the server. An application-supplied Transfer-Encoding is ignored. A single-body response gets one Content-Length; if the application supplied a length, it must match the body before any response bytes are written. Repeated or comma-joined lengths are accepted only when every numeric value is equal, then emitted as one canonical field.

On the native path a Response (or subclass) is serialised via Response.to_native(). A subclass that overrides __call__ to emit a custom event sequence is honoured on the WebSocket / external-host lanes but not on the native HTTP path — keep the wire behaviour in __call__ (shared by both lanes via the normalisers) or override to_native() to control the native serialisation.

`Response`¶

from blackbull import Response
from http import HTTPStatus

await send(Response(b'<h1>Hello</h1>'))
await send(Response('Hello', status=HTTPStatus.OK))           # str also accepted
await send(Response(b'Not found', status=HTTPStatus.NOT_FOUND))

# Redirect
await send(Response(b'', status=HTTPStatus.FOUND,
                    headers=[(b'location', b'/')]))

Default content_type is 'text/html; charset=utf-8'. Override via the content_type parameter:

Response(b'plain text', content_type='text/plain; charset=utf-8')

`JSONResponse`¶

from blackbull import JSONResponse

await send(JSONResponse({'ok': True}))
await send(JSONResponse({'error': 'Bad request'}, status=HTTPStatus.BAD_REQUEST))
await send(JSONResponse({'id': 1, 'title': 'Buy milk'}, status=HTTPStatus.CREATED))

Content-Type is set to application/json automatically.

`RedirectResponse`¶

from blackbull import RedirectResponse

await send(RedirectResponse('/new-url'))                                    # 302 Found
await send(RedirectResponse('/permanent', status=HTTPStatus.MOVED_PERMANENTLY))  # 301
return RedirectResponse('/login', status=HTTPStatus.SEE_OTHER)             # 303

Sets the Location header from the URL and an empty body. The default status is 302 Found — the safer general-purpose default, since it does not ask the client to preserve the original request method. The URL must be ASCII (RFC 9110 §10.2.2 — Location is a URI-reference); percent-encode non-ASCII URLs before passing them in. Extra headers= are merged in.

Custom response headers¶

Both Response and JSONResponse accept headers=[(bytes, bytes), ...]:

await send(JSONResponse({'ok': True}, headers=[
    (b'x-request-id', b'abc123'),
    (b'cache-control', b'no-store'),
]))

`Set-Cookie` helper¶

from blackbull import cookie_header

hdr = cookie_header('session', token, http_only=True)
# → (b'set-cookie', b'session=TOKEN; Path=/; HttpOnly; SameSite=Lax')

await send(JSONResponse({'ok': True}, headers=[hdr]))

Signature: cookie_header(name, value, path='/', http_only=True).

Cookies vs. tokens for SPA clients

Browsers may not reliably forward HttpOnly cookies set by a fetch() response on the next page navigation. For single-page apps, store the session token in sessionStorage and send it as Authorization: Bearer <token> instead.

HTTP trailers¶

HTTP/1.1 chunked responses can carry trailing headers after the body. Use the http.response.trailers event after the last http.response.body chunk:

@app.route(path='/chunked')
async def chunked(conn, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            (b'content-type',      b'text/plain'),
            (b'trailer',           b'x-checksum'),
        ],
        'trailers': True,
    })
    await send({
        'type': 'http.response.body',
        'body': b'chunk data here',
        'more_body': False,
    })
    await send({
        'type': 'http.response.trailers',
        'headers': [(b'x-checksum', b'abc123')],
        'more_trailers': False,
    })

trailers=True transfers response-completion ownership from the last body event to the last trailer event. BlackBull selects HTTP/1.1 chunked framing, ignores a supplied Content-Length, and writes the zero chunk only when the trailer section begins. To split a trailer section across events, set more_trailers=True on every event except the last. HTTP/2 collects those parts into one trailing HEADERS block carrying END_STREAM.

HEAD responses never carry content or a trailer section. The response is complete after its header section even if the application declared trailers; later trailer events are ignored so they cannot become bytes in the next keep-alive response.

Informational responses and status 204 carry neither content nor framing fields. A 205 carries no content and is emitted with Content-Length: 0 so its boundary remains explicit on a persistent connection. A 304 also carries no content; an explicit, valid Content-Length is retained as selected-representation metadata, but one is not generated from an attempted response body. Application body events for these statuses are discarded.

WebSocket frames¶

For WebSocket routes (scheme=Scheme.websocket), use WebSocketResponse to dispatch on payload type automatically:

from blackbull import WebSocketResponse

await send(WebSocketResponse('hello'))           # str  → text frame
await send(WebSocketResponse(b'\x00\x01'))       # bytes → binary frame
await send(WebSocketResponse({'type': 'msg'}))   # other → JSON-serialised text frame

See WebSockets for the full WebSocket surface.

Streaming responses¶

Use StreamingResponse to push an async generator to the client without buffering the whole body in memory. The class calls send directly, so it works as a nested ASGI app:

import asyncio
from blackbull import BlackBull, StreamingResponse

app = BlackBull()

async def countdown():
    for i in range(5, 0, -1):
        yield f'{i}\n'.encode()
        await asyncio.sleep(1)
    yield b'done\n'

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

StreamingResponse.__init__ accepts:

parameter	default	description
`content`	—	`AsyncIterator` of `bytes` or `str` chunks
`status`	`200`	HTTP status code
`headers`	`[]`	extra header tuples `(bytes, bytes)`
`media_type`	`'text/plain'`	`Content-Type` value (injected if absent from `headers`)

str chunks are encoded to UTF-8 automatically.

How HTTP/1.1 delivers streaming responses¶

When more_body=True appears on the first http.response.body event without a declared length, the HTTP/1.1 sender adds Transfer-Encoding: chunked and formats each body event as a hex-length chunk:

5\r\n
hello\r\n
5\r\n
world\r\n
0\r\n
\r\n

The terminal 0\r\n\r\n is written automatically when more_body=False arrives. When trailers=True was set in http.response.start, the final body event remains unterminated and the last http.response.trailers event writes 0\r\n, the trailer fields, and the final blank line.

A stream with a valid Content-Length keeps that single field and sends raw body bytes rather than chunk syntax. BlackBull checks the accumulated body size: crossing the declared boundary or ending short raises an error and the connection is not reused. Returning from an application after a non-terminal body event likewise leaves the response incomplete and closes that keep-alive connection.

HTTP/2 uses explicitly sized DATA frames instead of chunk syntax. Without trailers, END_STREAM maps to more_body=False; with declared trailers it belongs to the trailing HEADERS block.

Writing streaming-safe middleware¶

Any middleware that wraps the send callable and collects body parts will silently buffer a streaming response, defeating more_body=True.

For function-based middleware, the safest approach is to pass body events through immediately rather than accumulating them:

from blackbull.native import NativeResponse

async def prefix_mw(conn, receive, send, call_next):
    captured_start = None

    async def capturing_send(event):
        nonlocal captured_start
        if isinstance(event, NativeResponse):
            # BlackBull's own HTTP server (H1 + H2) threads native response
            # objects — see the middleware guide for the contract.  A single
            # object may carry header + body together.
            if event.body is not None and not event.more_body:
                # Non-streaming — transform the terminal body
                await send(NativeResponse(status=event.status,
                                          header=(list(event.header)
                                                  if event.header is not None
                                                  else None),
                                          body=b'[prefix] ' + event.body))
            else:
                # Streaming (or header-only) — pass through without buffering
                await send(event)
        elif event.get('type') == 'http.response.start':
            captured_start = event
        elif event.get('type') == 'http.response.body':
            if event.get('more_body'):
                # Streaming response — pass through without buffering
                await send(captured_start)
                captured_start = None
                await send(event)
            else:
                # Non-streaming — transform the body
                body = b'[prefix] ' + event.get('body', b'')
                await send(captured_start)
                await send({**event, 'body': body})
        else:
            await send(event)

    await call_next(conn, receive, capturing_send)

The two branches are the same policy on two wire shapes: the native object arm transforms only terminal bodies (so more_body=True chunks stream through untouched), and the dict arm keeps the original start/body sequencing for the WebSocket / external-host lanes (where the wire contract stays ASGI).

For most use cases (header injection, logging) the middleware does not touch the body at all and streaming safety is not a concern.

Detecting client disconnection¶

When the remote side closes the connection, receive() returns {'type': 'http.disconnect'}. This is useful for long-polling and server-sent events (SSE) — your handler can check whether the client is still there before writing more.

@app.route(path='/events')
async def sse(conn, receive, send):
    await send({
        'type': 'http.response.start',
        'status': 200,
        'headers': [
            (b'content-type',  b'text/event-stream'),
            (b'cache-control', b'no-cache'),
        ],
    })
    while True:
        event = await receive()
        if event['type'] == 'http.disconnect':
            break
        await send({
            'type': 'http.response.body',
            'body': b'data: ping\n\n',
            'more_body': True,
        })
    await send({'type': 'http.response.body', 'body': b'', 'more_body': False})

For most streaming use cases, prefer StreamingResponse (above) — it wraps an async generator and emits more_body=True on every chunk automatically. The raw event pattern is convenient when each chunk needs custom shaping (e.g. SSE event framing).

Passing Response(b'') to send ends the connection immediately with Content-Length: 0 and cannot be followed by additional chunks.

Next¶

Routing — @app.route, path parameters, route groups.
Middleware — wrapping handlers with cross-cutting concerns.
Static files — serving file-system content under a URL prefix.