Your First App¶

Hello World uses (conn, receive, send). A handler is full-form only when both receive and send are present; otherwise BlackBull injects its declared inputs and sends its return value.

Simplified handlers¶

Drop the parameters you don't use; return the response value instead of calling send:

from blackbull import BlackBull

app = BlackBull()

@app.route(path='/')
async def hello():
    return "Hello, world!"

BlackBull wraps the return value into a response automatically. No send call, no Response object, no scope boilerplate.

Path parameters¶

Declare parameters whose names match {name} segments in the path; BlackBull injects the captured value:

@app.route(path='/tasks/{task_id}')
async def get_task(task_id):       # str by default
    return f"Task {task_id}"

Add a type annotation and BlackBull will coerce the captured string for you:

@app.route(path='/tasks/{task_id}')
async def get_task(task_id: int):  # str captured, int() applied
    return {"id": task_id}

For URL patterns that should only match a specific type (so /tasks/abc returns 404 instead of being routed in and then failing to convert), use a path converter:

@app.route(path='/tasks/{task_id:int}')
async def get_task(task_id: int):  # router enforces int → no 500s on bad input
    return {"id": task_id}

Request body¶

Name a parameter body to receive the complete request body as bytes. BlackBull reads all chunks before calling the handler:

import json
from http import HTTPMethod

@app.route(path='/echo', methods=[HTTPMethod.POST])
async def echo(body: bytes):
    data = json.loads(body)
    return data          # dict → JSONResponse automatically

The `Connection` object¶

Annotate a parameter with Connection (or just name it conn/request) to receive a context object bundling the common reads — headers, cookies, client address, and the body helpers:

from blackbull import Connection

@app.route(path='/items/{item_id:int}', methods=[HTTPMethod.POST])
async def update_item(item_id: int, conn: Connection):
    lang = conn.headers.get(b'accept-language', b'en').decode()
    data = await conn.json()
    return {"id": item_id, "lang": lang, "data": data}

Request is the old name

Before v0.60.0 this object was called Request. blackbull.Request still works as a deprecated alias (BBD004); new code should use Connection.

See Requests and responses for the full surface.

The `conn` escape hatch¶

Name a parameter conn to receive the full Connection alongside other simplified parameters:

@app.route(path='/items/{item_id:int}')
async def get_item(item_id: int, conn):
    lang = conn.headers.get(b'accept-language', b'en').decode()
    return {"id": item_id, "lang": lang}

connection works as a synonym. A Connection is an object with typed attributes — conn.headers, conn.query_string, conn.cookies, conn.state, conn.extensions — not a mapping, so conn['headers'] and conn.get(...) do not work.

scope is not a valid parameter name

A simplified handler may not name a parameter scope; doing so raises TypeError when the route is registered. It was accepted until v0.65.0, when it injected the Connection — the name promised an ASGI scope dict and delivered something else, so scope['headers'] compiled fine and failed at request time. Rename it to conn.

Full-form handlers are unaffected: async def h(scope, receive, send) takes its arguments positionally, so the name is never inspected. It still receives a Connection, so conn reads better there too.

Return value mapping¶

What you return determines what BlackBull sends:

Return type	Response sent	Content-Type
`str`	`Response(value.encode())`	`text/html; charset=utf-8`
`bytes`	`Response(value)`	`text/html; charset=utf-8`
`dict` or `list`	`JSONResponse(value)`	`application/json`
`Response` / `JSONResponse`	Passed through as-is	(whatever the response sets)
`None`	Nothing sent	Handler called `send` directly, or intentionally empty

For finer control — custom status codes, extra headers, streaming bodies — return a Response / JSONResponse / StreamingResponse explicitly. See the Guide for details.

A small worked example¶

A tiny in-memory task tracker:

from http import HTTPMethod, HTTPStatus
from blackbull import BlackBull, Response

app = BlackBull()
_tasks: dict[int, dict] = {}
_next_id = 1


@app.route(path='/tasks', methods=[HTTPMethod.GET])
async def list_tasks():
    return list(_tasks.values())                # list → JSONResponse


@app.route(path='/tasks', methods=[HTTPMethod.POST])
async def create_task(body: bytes):
    global _next_id
    import json
    payload = json.loads(body)
    task = {'id': _next_id, 'title': payload['title'], 'done': False}
    _tasks[_next_id] = task
    _next_id += 1
    return task                                  # dict → JSONResponse


@app.route(path='/tasks/{task_id:int}', methods=[HTTPMethod.GET])
async def get_task(task_id: int):
    task = _tasks.get(task_id)
    if task is None:
        return Response(b'not found', status=HTTPStatus.NOT_FOUND,
                        content_type='text/plain')
    return task


if __name__ == '__main__':
    app.run(port=8000)

Run it:

$ python tasks.py
$ curl -X POST -d '{"title":"buy milk"}' localhost:8000/tasks
{"id": 1, "title": "buy milk", "done": false}

$ curl localhost:8000/tasks/1
{"id": 1, "title": "buy milk", "done": false}

$ curl localhost:8000/tasks
[{"id": 1, "title": "buy milk", "done": false}]

When to use the full triplet¶

The simplified form covers most needs, but not all:

WebSocket full-form handlers use (conn, receive, send); simplified handlers can inject a WebSocket helper. See WebSockets.
Middleware functions keep the (conn, receive, send, call_next) shape; the simplified adaptation does not apply to them.
Streaming uploads / long polling need to call receive() in a loop, so the full form is clearer.

The Guide covers each of these in turn.

Next¶

Guide — routing, middleware, WebSockets, HTTP/2, static files, error handling, configuration, logging, and more.