Middleware, health and rendering
Router and API middleware, health endpoints, fast JSON renderers and OpenTelemetry.
See Middleware and operations.
use_middleware()
def use_middleware(target: NinjaAPI | Router, *middlewares: Middleware) -> None: ...
Apply middlewares to every operation of an API or router (call before mounting).
| Parameter |
Type |
Default |
Description |
target |
NinjaAPI \| Router |
— |
A NinjaAPI or a Router. |
*middlewares |
Middleware |
— |
Middleware instances; the first one is outermost. |
middleware_decorator()
def middleware_decorator(*middlewares: Middleware) -> Callable[[Run], Run]: ...
A Ninja mode="view" decorator running middlewares (first is outermost).
| Parameter |
Type |
Default |
Description |
*middlewares |
Middleware |
— |
Middleware instances; the first one is outermost. |
RequestIDMiddleware
Accept or create a request id, expose it to the request and echo it in the response.
| Name |
Type |
Default |
Description |
header |
str |
'X-Request-ID' |
Header read from the request and written to the response. |
generate |
Callable[[], str] |
field(default=lambda: uuid.uuid4().hex) |
Makes an id when the client sends none. |
get_request_id()
def get_request_id(request: HttpRequest) -> str | None: ...
The id RequestIDMiddleware accepted or created for request.
| Parameter |
Type |
Default |
Description |
request |
HttpRequest |
— |
The current request. |
DeprecationMiddleware
Deprecation (RFC 9745), Sunset (RFC 8594) and Link headers for old versions.
| Name |
Type |
Default |
Description |
deprecated_at |
datetime \| None |
None |
When the API was deprecated; None sends Deprecation: true. |
sunset |
datetime \| None |
None |
When it stops working. |
link |
str \| None |
None |
Migration guide URL (Link: <...>; rel="deprecation"). |
Other middleware
| Class |
Arguments |
Description |
ServerTimingMiddleware |
— |
Server-Timing: app;dur=<ms> for browser dev tools and APM. |
RateLimitHeadersMiddleware |
— |
RateLimit-Limit/Remaining/Reset and RateLimit-Policy from ninja-devx throttles. |
SecurityHeadersMiddleware |
*, hsts: str \| None = None, csp: str \| None = None, referrer_policy: str = 'same-origin', frame_options: str \| None = 'DENY', permissions_policy: str \| None = _PERMISSIONS_POLICY, nosniff: bool = True, extra: Mapping[str, str] \| None = None |
Set security headers on every response unless the response already has them. |
ResponseCacheMiddleware |
*, ttl: int = 60, vary_on: Sequence[str] = (), cache: str = 'default', key_prefix: str = '', methods: Sequence[str] = ('GET', 'HEAD') |
Serve matching responses from the cache and store new ones. |
MaxBodySizeMiddleware |
max_bytes: int |
Reject requests whose Content-Length exceeds max_bytes with 413. |
EnforceContentTypeMiddleware |
media_types: Collection[str], *, methods: Collection[str] = _BODY_METHODS |
Reject write requests whose media type is not allowed, with 415. |
JsonDepthMiddleware |
max_depth: int = 32 |
Reject JSON bodies nested deeper than max_depth with 400. |
PaginationHeadersMiddleware |
— |
Copy pagination metadata recorded by the paginator onto response headers. |
OpenTelemetryMetricsMiddleware |
meter: Meter \| None = None |
HTTP server metrics per operation (OpenTelemetry semantic conventions). |
invalidate_cache()
def invalidate_cache(prefix: str = '', *, cache: str = 'default') -> None: ...
Invalidate every cached response under prefix.
| Parameter |
Type |
Default |
Description |
prefix |
str |
'' |
The same prefix passed to ResponseCacheMiddleware. |
cache |
str |
'default' |
Cache alias. |
record_rate_limit()
def record_rate_limit(request: HttpRequest, *, policy: str, limit: int, remaining: int, reset: float, window: int) -> None: ...
Remember a throttle's state for RateLimitHeadersMiddleware (throttles call it).
| Parameter |
Type |
Default |
Description |
request |
HttpRequest |
— |
The current request. |
policy |
str |
— |
Throttle name, shown in RateLimit-Policy. |
limit |
int |
— |
Requests allowed per window. |
remaining |
int |
— |
Requests left in the current window. |
reset |
float |
— |
Seconds until the window resets. |
window |
int |
— |
Window length in seconds. |
QueryExplainMiddleware()
def QueryExplainMiddleware(*, enabled: bool | None = None, databases: Sequence[str] = ()) -> None: ...
Add query diagnostics headers to matching responses.
| Parameter |
Type |
Default |
Description |
enabled |
bool \| None |
None |
Forces the middleware on or off; None (the default) follows settings.DEBUG on every request, so override_settings(DEBUG=...) works. |
databases |
Sequence[str] |
() |
Database aliases to count; defaults to every configured alias. |
VersionedResponseMixin
Adds Accept-Version negotiation to every operation of a controller.
| Name |
Type |
Default |
Description |
response_versions |
Mapping[int, type[BaseModel]] |
MappingProxyType({}) |
Older response schemas, keyed by the version clients ask for with Accept-Version. The latest version is implicitly one more than the highest key here. |
response_version_header |
str |
'Accept-Version' |
Request header naming the wanted version. |
response_version_response_header |
str |
'X-API-Version' |
Response header naming the version actually served. |
VersionedResponseMiddleware()
def VersionedResponseMiddleware(response_versions: Mapping[int, type[BaseModel]], *, header: str = 'Accept-Version', response_header: str = 'X-API-Version') -> None: ...
Negotiate Accept-Version and downgrade the rendered response.
| Parameter |
Type |
Default |
Description |
response_versions |
Mapping[int, type[BaseModel]] |
— |
{version: schema} for every version but the latest. |
header |
str |
'Accept-Version' |
Request header naming the wanted version. |
response_header |
str |
'X-API-Version' |
Response header naming the version actually served. |
RequestLogMiddleware
Logs one record per request. Install directly, or through RequestLogPlugin.
| Name |
Type |
Default |
Description |
logger |
logging.Logger |
field(default_factory=lambda: logging.getLogger('ninja_devx.request')) |
Logger receiving one record per request. |
level |
int |
logging.INFO |
Log level of the record. |
naming |
Naming |
'otel' |
"otel": OpenTelemetry semantic convention field names; "flat": plain ones. |
bind_structlog |
bool |
True |
Bind the identity fields to structlog.contextvars when structlog is installed. |
RequestLogPlugin
RequestLogMiddleware plus RequestIDMiddleware, so every request gets an id.
| Name |
Type |
Default |
Description |
logger |
logging.Logger |
field(default_factory=lambda: logging.getLogger('ninja_devx.request')) |
Logger receiving one record per request. |
level |
int |
logging.INFO |
Log level of the record. |
naming |
Naming |
'otel' |
"otel": OpenTelemetry semantic convention field names; "flat": plain ones. |
bind_structlog |
bool |
True |
Bind the identity fields to structlog.contextvars when structlog is installed. |
include_request_id |
bool |
True |
Add RequestIDMiddleware too, so request_id is never empty. |
register_api_key_reader()
def register_api_key_reader(reader: ApiKeyReader) -> None: ...
Let ninja_devx.contrib.apikeys (or your own auth) contribute api_key_prefix.
| Parameter |
Type |
Default |
Description |
reader |
ApiKeyReader |
— |
Returns the authenticating key's prefix, or None. |
| Class |
Arguments |
Description |
JSONFormatter |
— |
Render each LogRecord as one JSON object, extra fields included. |
HealthController
GET /live and GET /ready.
| Name |
Type |
Default |
Description |
health_checks |
Sequence[HealthCheck] |
(DatabaseCheck(),) |
Checks run concurrently by /ready; results retain this order. |
health_timeout |
float |
2.0 |
Total readiness deadline in seconds. Hung checks retain their bounded worker slot. |
Health checks
| Class |
Arguments |
Description |
DatabaseCheck |
alias: str = DEFAULT_DB_ALIAS, name: str = 'database' |
Runs SELECT 1 on a database connection. |
CacheCheck |
alias: str = 'default', name: str = 'cache' |
Writes and reads a key in a cache. |
MigrationsCheck |
alias: str = DEFAULT_DB_ALIAS, name: str = 'migrations' |
Fails while migrations are not applied (useful during deploys). |
ORJSONRenderer
JSON with orjson (ninja-devx[orjson]): NinjaAPI(renderer=ORJSONRenderer()).
| Name |
Type |
Default |
Description |
options |
int \| None |
None |
Extra orjson.OPT_* flags (orjson.OPT_INDENT_2...). |
Other renderers
| Class |
Arguments |
Description |
MsgspecRenderer |
— |
JSON with msgspec (ninja-devx[msgspec]); same output rules as ORJSONRenderer. |
OpenTelemetryHook
| Name |
Type |
Default |
Description |
tracer |
Tracer \| None |
None |
Tracer starting one span per call (default: the global tracer). |