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

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

blackbull.response¶