`blackbull.response`¶

`blackbull.response` ¶

HTTP and WebSocket response objects.

Provides:

Response: plain HTTP response (HTML / plain text / binary).
JSONResponse: convenience subclass that serialises a Python object to JSON.
RedirectResponse: convenience subclass that sets a Location header + 3xx status.
StreamingResponse: pushes an async iterator to the client without buffering.
EventSourceResponse: WHATWG Server-Sent Events on top of StreamingResponse.
WebSocketResponse: wraps text, bytes, or dict data as a WebSocket send event.
cookie_header: builds a (b'set-cookie', ...) header tuple with secure defaults.

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

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

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

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

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

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

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

`wrap_native_send(raw_send)` ¶

Handler-facing send adapter: every accepted shape → NativeResponse.

The native-ization flip of app._wrap_send: instead of normalising convenience shapes (Response, 3-arg) into ASGI dicts so downstream sees plain dicts, every accepted shape becomes a :class:~blackbull.native.NativeResponse here, so everything above the route handler (route-header injection, middleware, access log, sender) observes a single native representation on the H1 path.

Shared by app._wrap_send_native (the handler boundary) and middleware.utils._normalize_send (as_middleware's call_next), so global and per-route middleware see the same native contract.

Accepted shapes (full-form send — the compat contract, held until 2027-07-29):

Response (incl. subclasses) — one NativeResponse (to_native());
StreamingResponse / EventSourceResponse — driven; their own start/body dict sequence converts per-event below;
(bytes, status, headers) 3-arg form — one NativeResponse;
ASGI dicts — per-event: http.response.start → header arm (the trailers flag → expects_trailers), http.response.body → body chunk, http.response.trailers → trailers arm; push / pathsend / disconnect / unknown pass through (the bilingual sender decides — push is H2-only, pathsend is the static middleware's deferred form);
NativeResponse — pass through;
anything else — pass through so the sender's type check decides.

blackbull.response¶