`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 expanded into ASGI event dicts before reaching the middleware's inner send wrapper. The wrapper therefore only ever sees plain dict events and does not need isinstance guards.

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 always a dict here
        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 always a dict here
            ...
        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¶