`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(scope, receive, send): ...

`Session` ¶

ASGI middleware that maintains a signed-cookie session.

Parameters¶

secret: The HMAC key. Bytes or str. When omitted, BB_SESSION_SECRET is read from the environment. A missing / empty secret raises at construction — no insecure default. cookie_name: Name of the cookie carrying the session payload. Default 'session'. max_age: Cookie Max-Age in seconds. When set, the cookie is signed with a server-side timestamp; values older than max_age seconds are treated as expired (empty session). None means a session cookie that lives only as long as the browser is open. secure: Set the Secure attribute so the cookie is only sent over HTTPS. Default True; set False for local-only dev. httponly: Set the HttpOnly attribute (JavaScript can't read the cookie). Default True. samesite: Strict / Lax / None. Default 'Lax'. path: Cookie Path. Default /.

`StaticFiles` ¶

`TrustedProxy` ¶

Rewrite scope['client'] and scope['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 scope['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 (scope, receive, send, call_next))::

@as_middleware
async def timing_mw(scope, receive, send, call_next):
    async def timed_send(event):
        # event is always a dict here
        await send(event)
    await call_next(scope, receive, timed_send)

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

@as_middleware
class Cache:
    async def __call__(self, scope, receive, send, call_next):
        async def cap_send(event):
            # event is always a dict here
            ...
        await call_next(scope, 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¶