`blackbull.middleware`¶

`blackbull.middleware` ¶

Public middleware exports.

Names are short nouns by convention — the module path (blackbull.middleware) supplies the "this is middleware" context, so the type names don't need a redundant suffix. This matches the project's earliest middlewares (CORS, StaticFiles).

Deprecated aliases for the previous *Middleware-suffixed names and the compress pre-built instance are kept available through PEP 562 __getattr__ so existing user code keeps working with a one-time DeprecationWarning. They will be removed in a future release.

`CORS` ¶

Cross-Origin Resource Sharing (CORS) middleware.

Handles preflight OPTIONS requests and attaches CORS headers to actual cross-origin responses. Requests without an Origin header pass through unchanged.

Parameters:

Name	Type	Description	Default
`allow_origins`	`list[str] \| str`	Explicit origin strings or `['*']` for wildcard.	`'*'`
`allow_methods`	`list[str] \| None`	HTTP methods permitted in cross-origin requests. Defaults to `['GET', 'POST', 'HEAD', 'OPTIONS']`.	`None`
`allow_headers`	`list[str] \| str`	Request headers permitted; `['*']` allows all.	`'*'`
`allow_credentials`	`bool`	Emit `Access-Control-Allow-Credentials: true`. Cannot be combined with `allow_origins=['*']`.	`False`
`expose_headers`	`list[str] \| None`	Response headers the browser JS may read.	`None`
`max_age`	`int \| None`	Preflight cache lifetime in seconds. `None` omits the header.	`600`

Usage::

app = BlackBull()
app.use(CORS(
    allow_origins=['https://myapp.example.com'],
    allow_credentials=True,
    max_age=3600,
))

`Cache` ¶

Per-worker in-memory response cache.

`Compression` ¶

ASGI middleware: compress the response body using the best codec the client accepts (br > zstd > gzip, in server-preference order).

Bodies smaller than min_size bytes are forwarded uncompressed. Responses with already-compressed Content-Types (image/, video/, etc.) are forwarded uncompressed. brotli and zstandard are optional — if not installed the middleware falls back gracefully to gzip or no compression.

BlackBull middleware convention::

from blackbull.middleware import Compression

@app.route(path='/', middlewares=[Compression()])
async def handler(conn, receive, send): ...

`StaticFiles` ¶

`init(directory=None, *, url_prefix='', root_dir=None, cache=False, index=None, conditional=True)` ¶

Serve files from directory (or root_dir).

index (default None — off): when set to a filename (e.g. 'index.html'), a request that resolves to a directory is served that file from inside the directory if it exists. Off by default so existing exact-path serving is unchanged; the blackbull serve CLI turns it on to match python -m http.server's directory-index behaviour.

cache (default False): when True, file bodies up to _CACHE_MAX_BYTES_PER_FILE are held in an in-memory OrderedDict (capped at _CACHE_MAX_ENTRIES), and the per-request stat() syscall is throttled by _STAT_TTL_S. When False (the default), every request does a fresh stat() and reads the body from disk — matching the behaviour of Starlette / FastAPI / Flask static serving and the requirement HttpArena's standard-mode rules place on static profiles ("read files from disk on every request, no in-memory caching"). Set cache=True only for standalone deployments where BlackBull terminates static traffic directly (i.e. no nginx / CDN in front).

conditional (default True): emit ETag + Last-Modified validators and honour If-None-Match / If-Modified-Since with a 304. Set False to suppress validators (e.g. the blackbull serve --no-etag path).

`TrustedProxy` ¶

Rewrite conn['client'] and conn['scheme'] from proxy headers.

Applied only when the direct TCP peer matches the configured trusted set, preventing malicious clients from spoofing X-Forwarded-For.

Supported headers (in precedence order):

RFC 7239 Forwarded — for=<ip>; proto=<scheme>
X-Forwarded-For — comma-separated IP chain; leftmost non-trusted IP wins
X-Forwarded-Proto — rewrite conn['scheme']

Parameters:

Name	Type	Description	Default
`trusted_proxies`	`list[str] \| str \| None`	IP addresses or CIDR strings (IPv4 or IPv6). Accepts a single string or a list. Defaults to loopback (`'127.0.0.1'`, `'::1'`).	`None`

Usage::

app = BlackBull(trusted_proxies=['127.0.0.1', '10.0.0.0/8'])

# or register explicitly for more control:
from blackbull import TrustedProxyMiddleware
app.use(TrustedProxyMiddleware(['127.0.0.1', '::1']))

`as_middleware(target)` ¶

Decorator that marks an async function or class as BlackBull middleware.

Wraps call_next so any send callable the middleware passes to it is automatically normalised — Response/JSONResponse objects are converted to NativeResponse before reaching the middleware's inner send wrapper. The wrapper therefore only ever sees the native representation (NativeResponse on the HTTP path — H1 and H2).

Applied to an async function (signature (conn, receive, send, call_next))::

@as_middleware
async def timing_mw(conn, receive, send, call_next):
    async def timed_send(event):
        # event is a NativeResponse on the HTTP path
        await send(event)
    await call_next(conn, receive, timed_send)

Applied to a class whose __call__ is the middleware coroutine::

@as_middleware
class Cache:
    async def __call__(self, conn, receive, send, call_next):
        async def cap_send(event):
            # event is a NativeResponse on the HTTP path
            ...
        await call_next(conn, receive, cap_send)

Power users who need to handle raw send arguments (e.g. because their middleware is used in a context where no simplified handlers are registered) should omit this decorator — their call_next is then wired directly to the next handler with no extra wrapping.

blackbull.middleware¶