`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.
StreamingResponse: pushes an async generator to the client without buffering.
WebSocketResponse: wraps text, bytes, or dict data as a WebSocket send event.
cookie_header: builds a (b'set-cookie', ...) header tuple with secure defaults.

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

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

`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(scope, receive, send):
    await StreamingResponse(lines())(scope, 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¶