API keys, audit log, webhooks, uploads and jobs

The optional Django apps and adapters in ninja_devx.contrib. The apps (apikeys, audit, webhooks, uploads, jobs) need INSTALLED_APPS and a migration; redis_throttle and nplusone are adapters with no app of their own. See API keys, Audit log, Webhooks, Uploads, Jobs, Operations and N+1 detection and devx_doctor.

API keys (ninja_devx.contrib.apikeys)

create_api_key()

def create_api_key(user: object, name: str, *, scopes: Iterable[str] = (), expires_at: datetime | None = None, rate_limit: str = '') -> tuple[APIKey, str]: ...

Create a key and return it with its raw value, which is never stored or shown again.

Parameter	Type	Default	Description
user	object	—	The key's owner; requests authenticated with it act as this user.
name	str	—	A label for humans ("CI deploys").
scopes	Iterable[str]	()	What the key may do ("orders:read", "orders:*", "*").
expires_at	datetime \| None	None	When the key stops working (None: never).
rate_limit	str	''	This key's rate for APIKeyRateThrottle ("1000/hour"; empty: the throttle's default).

revoke_api_key()

def revoke_api_key(key: APIKey) -> None: ...

Stop accepting key (kept for audit).

Parameter	Type	Default	Description
key	APIKey	—	The key to revoke.

Authentication and scopes

Class	Arguments	Description
APIKeyAuth	—	X-API-Key: ndx_...; request.auth is the key's user.
APIKeyBearer	—	Authorization: Bearer ndx_....
RequiresScope	scope: str	Operation metadata: meta=(RequiresScope("orders:write"),).
HasScopes	allow_unscoped: bool = False	Every RequiresScope of the operation must be granted to the request's API key.
APIKeyRateThrottle	—	Per API key: the key's own rate_limit, else rate; other requests pass.

scope_allows()

def scope_allows(granted: Sequence[str], required: str) -> bool: ...

Whether granted covers required: exact, "orders:*" or "*".

Parameter	Type	Default	Description
granted	Sequence[str]	—	Scopes of the key.
required	str	—	The scope an operation needs ("orders:write").

APIKeyController

GET / my keys, POST / create one (raw key in the response), DELETE /{id}.

Name	Type	Default	Description
grantable_scopes	Sequence[str] \| None	None	Scopes users may put on their keys (None: any).

Audit log (ninja_devx.contrib.audit)

AuditMixin

Record creates, updates and deletes of a model controller (bulk ones too).

Name	Type	Default	Description
audit_privacy	AuditPrivacy	AuditPrivacy()	Storage allowlists, credential redaction and safe object labels.
audit_fields	Sequence[str] \| None	None	Fields kept in snapshots and diffs (None: every concrete field).
audit_exclude	Sequence[str]	('password',)	Fields never recorded.
audit_redact	Sequence[str]	()	Fields recorded as "***": a change is visible, the value is not.

AuditPrivacy

Allowlist business fields and redact credential values before persistence.

Name	Type	Default	Description
fields	tuple[str, ...] \| None	None	Allowlist of change field names to keep (None: keep every field not excluded).
exclude	tuple[str, ...]	('password', 'hashed_secret', 'private_key')	Change field names dropped entirely, never stored even redacted.
redact	tuple[str, ...]	('secret', 'token', 'access_token', 'refresh_token', 'api_key', 'authorization')	Change field names stored as "***" instead of their value.
metadata_fields	tuple[str, ...] \| None	None	Allowlist of metadata keys to keep (None: keep every key).
object_repr	bool	False	Opt in to calling the model's potentially sensitive __str__.
schema	type[object] \| None	None	A schema (ninja_devx.serialization.privacy.Sensitive-annotated) whose marked fields are redacted like redact names.

record()

def record(request: HttpRequest | None, action: str, obj: models.Model | None = None, *, object_pk: object = None, changes: Mapping[str, object] | None = None, metadata: Mapping[str, object] | None = None, using: str | None = None, privacy: AuditPrivacy | None = None) -> AuditEntry: ...

Write an audit entry for action on obj, attributed to request's user.

Parameter	Type	Default	Description
request	HttpRequest \| None	—	The request (actor, request id, method, path, IP); None for jobs.
action	str	—	A short verb: "create", "update", "delete", "export"...
obj	models.Model \| None	None	The object acted on, if any.
object_pk	object	None	The key when obj no longer has one (after a delete).
changes	Mapping[str, object] \| None	None	{field: [old, new]}.
metadata	Mapping[str, object] \| None	None	Anything else worth keeping (JSON-serializable).
using	str \| None	None	Database alias; defaults to the object database, then the audit write router.
privacy	AuditPrivacy \| None	None	Storage allowlists/redaction policy; default protects common credential fields.

snapshot()

def snapshot(instance: models.Model, *, fields: Sequence[str] | None = None, exclude: Sequence[str] = (), redact: Sequence[str] = ()) -> dict[str, object]: ...

JSON-safe values of instance's concrete fields (foreign keys as their key).

Parameter	Type	Default	Description
instance	models.Model	—	The model instance.
fields	Sequence[str] \| None	None	Field names to keep (None: all concrete fields).
exclude	Sequence[str]	()	Field names to leave out.
redact	Sequence[str]	()	Field names whose values become "***".

diff()

def diff(before: Mapping[str, object], after: Mapping[str, object]) -> dict[str, list[object]]: ...

{field: [old, new]} for the fields whose value changed.

Parameter	Type	Default	Description
before	Mapping[str, object]	—	A snapshot taken before the change.
after	Mapping[str, object]	—	A snapshot taken after it.

redact()

def redact(changes: Mapping[str, list[object]], fields: Sequence[str]) -> dict[str, list[object]]: ...

Replace the values of fields by "***" (a change stays visible).

Parameter	Type	Default	Description
changes	Mapping[str, list[object]]	—	{field: [old, new]}.
fields	Sequence[str]	—	Field names to hide.

AuditFilters

Query parameters of GET / on AuditLogController.

Name	Type	Default	Description
action	str \| None	None	create, update, delete or a custom action.
actor_id	int \| str \| UUID \| None	None	Entries by this user.
model	str \| None	Field(None, description='app_label.model.')	Entries about this model: app_label.model.
object_pk	str \| None	None	Entries about this object (with model).
request_id	str \| None	None	Everything recorded during one request.
since	Annotated[datetime \| None, FilterLookup('created__gte')]	None	Entries at or after this time.
until	Annotated[datetime \| None, FilterLookup('created__lt')]	None	Entries before this time.

Webhooks (ninja_devx.contrib.webhooks)

publish()

def publish(event_type: str, payload: Mapping[str, object], *, owner: Model | None = None, tenant_key: str = '', broadcast: bool = False, using: str | None = None, queue: TaskQueue | None = None, task: Enqueueable[[str, str]] | None = None) -> OutboxEvent: ...

Store an event and a pending delivery per subscribed endpoint, in the current transaction (DjangoJSONEncoder serializes dates, decimals and UUIDs).

Parameter	Type	Default	Description
event_type	str	—	Dotted event name ("order.created"); endpoints subscribe to it.
payload	Mapping[str, object]	—	The event data, sent as data in the request body.
owner	Model \| None	None	Only this user's endpoints, within tenant_key.
tenant_key	str	''	Server-controlled tenant scope; empty means personal endpoints.
broadcast	bool	False	Explicitly send to every subscribed endpoint; cannot combine with a target.
using	str \| None	None	Database alias shared by business writes and the outbox.
queue	TaskQueue \| None	None	Where to enqueue the delivery (OnCommitTaskQueue()).
task	Enqueueable[[str, str]] \| None	None	The task receiving the event id and database alias. Default: deliver_event_task (Django 6.0+); pass your Celery/RQ wrapper otherwise.

deliver_due()

def deliver_due(*, limit: int = 100, transport: Transport | None = None, timeout: float = 10.0, schedule: Sequence[timedelta] = RETRY_SCHEDULE, disable_after: timedelta | None = timedelta(days=5), url_policy: URLPolicy | None = None, event_id: str | None = None, now: datetime | None = None, using: str | None = None) -> DeliveryReport: ...

Send up to limit due deliveries. Safe to run from several workers (each row uses a conditional claim and a token-checked finalization).

Parameter	Type	Default	Description
limit	int	100	Deliveries handled in this call.
transport	Transport \| None	None	How requests are sent (default SafeHTTPTransport(url_policy)).
timeout	float	10.0	Seconds per request.
schedule	Sequence[timedelta]	RETRY_SCHEDULE	Delays before each retry; the delivery fails after the last one.
disable_after	timedelta \| None	timedelta(days=5)	Disable an endpoint failing continuously for this long.
url_policy	URLPolicy \| None	None	Where the default transport may send requests.
event_id	str \| None	None	Only deliveries of this event.
now	datetime \| None	None	The current time (tests).
using	str \| None	None	Database containing the outbox.

deliver_event()

def deliver_event(event_id: str, *, using: str | None = None) -> DeliveryReport: ...

Send the due deliveries of one event now (what enqueued tasks run).

Parameter	Type	Default	Description
event_id	str	—	The OutboxEvent primary key.
using	str \| None	None	Database alias containing the outbox event.

event_matches()

def event_matches(patterns: Sequence[str], event_type: str) -> bool: ...

"*", an exact type, or a "order.*" prefix.

Parameter	Type	Default	Description
patterns	Sequence[str]	—	An endpoint's subscriptions.
event_type	str	—	The published event type.

verify_signature()

def verify_signature(secrets_: str | list[str], headers: Mapping[str, str], body: bytes, *, tolerance: int = 300, now: float | None = None) -> None: ...

Raise InvalidSignature unless body was signed by one of secrets_.

Parameter	Type	Default	Description
secrets_	str \| list[str]	—	The endpoint secret, or several during a rotation.
headers	Mapping[str, str]	—	Request headers (any case); request.headers works.
body	bytes	—	The raw request body (request.body), not re-serialized JSON.
tolerance	int	300	Maximum age of the timestamp, in seconds.
now	float \| None	None	The current Unix time (tests).

signature_headers()

def signature_headers(secret: str, message_id: str, body: bytes, *, timestamp: int | None = None) -> dict[str, str]: ...

The webhook-id, webhook-timestamp and webhook-signature headers.

Parameter	Type	Default	Description
secret	str	—	The endpoint secret (whsec_...).
message_id	str	—	Unique message id, the same across retries.
body	bytes	—	The exact bytes that are sent.
timestamp	int \| None	None	Unix time of the attempt (default: now).

generate_secret()

def generate_secret() -> str: ...

A new signing secret: whsec_<base64 of 24 random bytes>.

URLPolicy

Where webhooks may be sent.

Name	Type	Default	Description
allow_http	bool	False	Accept http:// URLs (default: HTTPS only).
allow_private_networks	bool	False	Accept loopback, private, link-local and other non-public addresses (development).

check_url()

def check_url(url: str, policy: URLPolicy | None = None) -> None: ...

Raise UnsafeURL when url breaks policy (static checks, no DNS).

Parameter	Type	Default	Description
url	str	—	The endpoint URL.
policy	URLPolicy \| None	None	What is allowed.

SafeHTTPTransport

The default webhook transport: urllib, no redirects, no environment proxies, and policy enforced on the URL and on the connected address.

Name	Type	Default	Description
policy	URLPolicy	URLPolicy()	Where requests may go.

encrypt_secret()

def encrypt_secret(raw: str) -> str: ...

The value to store for raw: encrypted when keys are configured.

Parameter	Type	Default	Description
raw	str	—	The signing secret (whsec_...).

decrypt_secret()

def decrypt_secret(stored: str) -> str: ...

The raw secret from a stored value (plain values are returned as they are).

Parameter	Type	Default	Description
stored	str	—	The database value.

reencrypt_secret()

def reencrypt_secret(stored: str) -> str: ...

stored encrypted with the first key (plain values get encrypted).

Parameter	Type	Default	Description
stored	str	—	The database value.

WebhookEndpointController

The user's endpoints: CRUD, deliveries, retry, ping and secret rotation.

Name	Type	Default	Description
url_policy	URLPolicy	URLPolicy()	Which URLs may be registered (HTTPS, public addresses by default). Pass the same policy to the delivery worker.
available_events	Sequence[str] \| None	None	Event types clients may subscribe to (None: any); patterns must match one.

Uploads (ninja_devx.contrib.uploads)

UploadController

POST / signs an upload, POST /complete verifies it. Set signer.

Name	Type	Default	Description
signer	UploadSigner \| None	None	Storage backend (S3Signer, FakeSigner or your own).
policy	UploadPolicy	UploadPolicy()	Allowed types, size and expiry.
upload_database	str	'default'	Alias containing core UploadRecord rows (run migrations on it).
key_prefix	str	''	Added before <user id>/<uuid>/<filename> (after the signer's own prefix).

UploadPolicy

What may be uploaded.

Name	Type	Default	Description
content_types	Sequence[str]	('*/*',)	Allowed media types; "image/*" allows a family.
max_bytes	int	10 * 1024 * 1024	Largest accepted size.
expires_in	timedelta	timedelta(minutes=10)	How long the signed form stays valid.
cleanup_grace	timedelta	timedelta(minutes=15)	Grace after form expiry, allowing in-flight uploads to finish before cleanup.
require_checksum	bool	False	Require a base64 SHA-256 digest in the signed request and stored object.
require_version	bool	False	Require storage versioning; consumers must read the returned version_id.

S3Signer

Presigned POST uploads to S3 (or MinIO, R2...) with boto3 (ninja-devx[s3]).

Name	Type	Default	Description
bucket	str	—	Bucket name.
prefix	str	'uploads/'	Key prefix for every upload of this signer.
client	S3Client \| None	None	A boto3 S3 client (default: boto3.client("s3"), created on first use).

Test signer

Class	Arguments	Description
FakeSigner	url: str = 'https://storage.test/upload', objects: dict[str, StoredObject] = field(default_factory=dict[str, StoredObject]), signed: list[str] = field(default_factory=list[str])	In-memory signer for tests: signer.store(key, size, content_type) simulates the client's upload.

safe_filename()

def safe_filename(name: str, *, max_length: int = 100) -> str: ...

ASCII, no directories, no leading dots: "../Résumé 2024.pdf" → "Resume_2024.pdf".

Parameter	Type	Default	Description
name	str	—	The client's file name.
max_length	int	100	Longest result, extension included.

Jobs (ninja_devx.contrib.jobs)

JobsController: GET / (the caller's jobs, paginated), GET /{id} and POST /{id}/cancel. Mount next to the endpoints that call start_job. Maintain with manage.py devx_jobs prune/retry (see Commands).

JobOut

Name	Type	Default	Description
id	UUID	—	Primary key, also the value start_job/accepted return in the Location.
name	str	—	Recorded label, shown in the API and admin.
status	str	—	One of Job.Status: queued, running, succeeded, failed, cancelled.
progress	int	—	Last value reported through JobContext.set_progress (0 to 100).
result	dict[str, object]	—	The function's return value, once succeeded.
error	str	—	The failure message, once failed.
attempts	int	—	Times the job has started running, including retries.
created	datetime	—	When start_job created the row.
started	datetime \| None	—	When the job most recently started running.
finished	datetime \| None	—	When the job reached a terminal status.

start_job()

def start_job(request: HttpRequest, name: str, function: JobFunction | str, *args: object, queue: TaskQueue | None = None, tenant: str = '', **kwargs: object) -> Job: ...

Create a queued job for the caller and enqueue its run.

Parameter	Type	Default	Description
request	HttpRequest	—	The current request; created_by is the authenticated user (401 without one).
name	str	—	Recorded on the job; shown in the API and admin.
function	JobFunction \| str	—	A @job-registered function, its registered name, or a dotted import path. A plain function is recorded by module.qualname.
*args	object	—	Positional arguments passed to the function after its JobContext.
queue	TaskQueue \| None	None	Where run_job is enqueued. Default: a TaskQueue singleton from the request's container, else OnCommitTaskQueue().
tenant	str	''	Recorded on the job (server-controlled; not read from the request body).
**kwargs	object	—	Keyword arguments passed to the function.

run_job()

def run_job(job_id: str, target: str, retry: int = 0, /, *args: object, **kwargs: object) -> None: ...

Run a job's function inside a JobContext, storing the outcome on its row.

Parameter	Type	Default	Description
job_id	str	—	Primary key of the Job row (also passed to the JobContext).
target	str	—	A name registered with @job, or a dotted import path.
retry	int	0	Extra attempts on failure, made in this same call before giving up.
*args	object	—	Positional arguments passed to the function after its JobContext.
**kwargs	object	—	Keyword arguments passed to the function.

accepted()

def accepted(request: HttpRequest, row: Job, *, prefix: str = '/jobs') -> JsonResponse: ...

A 202 response pointing at the job: a Location header and a small body.

Parameter	Type	Default	Description
request	HttpRequest	—	The current request, used to build an absolute URL.
row	Job	—	The job just started.
prefix	str	'/jobs'	Where JobsController is mounted.

job()

def job(name: str) -> Callable[[JobFunction], JobFunction]: ...

Register a function so workers resolve it by name instead of pickling it.

Parameter	Type	Default	Description
name	str	—	The name start_job and run_job use to find the function again.

resolve_job()

def resolve_job(target: str) -> JobFunction: ...

The function registered, or importable, as target.

Parameter	Type	Default	Description
target	str	—	A name passed to @job, or a dotted import path.

JobContext()

def JobContext(job_id: str) -> None: ...

Given to a job function: progress reporting and cooperative cancellation.

Parameter	Type	Default	Description
job_id	str	—	Primary key of the Job row this context reports on.

Redis throttle storage (ninja_devx.contrib.redis_throttle)

Throttle storage

Class	Arguments	Description
RedisThrottleStorage	client: _RedisClient \| None = None, *, url: str = 'redis://localhost:6379/0'	ThrottleStorage counting fixed windows on Redis with one atomic round trip.

N+1 detection (ninja_devx.contrib.nplusone)

An adapter over django-zeal (the zeal extra): add NPlusOnePlugin()/NPlusOneMiddleware() so every request runs inside zeal's tracking context, and its errors are rewritten to name the controller and the fix. A no-op when django-zeal is not installed.

N+1 middleware

Class	Arguments	Description
NPlusOneMiddleware	—	Runs the request inside zeal's tracking context and explains its errors.
NPlusOnePlugin	—	Adds NPlusOneMiddleware to every operation of the API.

explain_n_plus_one()

def explain_n_plus_one(message: str, *, controller: str | None = None) -> str: ...

Rewrite a zeal NPlusOneError message to name controller and suggest a fix.

Parameter	Type	Default	Description
message	str	—	str(exc) from the NPlusOneError zeal raised.
controller	str \| None	None	Qualified name of the controller handling the request, if known.

zeal_installed()

def zeal_installed() -> bool: ...

Whether django-zeal is importable.

zeal_strict()

def zeal_strict() -> Generator[None]: ...

zeal.zeal_context() forced to raise, regardless of settings.ZEAL_RAISE.

django-rules (ninja_devx.contrib.rules)

Rules from django-rules (the rules extra) as permissions, checked on the request and on loaded objects (see Permissions).

Rule permissions

Class	Arguments	Description
HasRule	rule: str \| Predicate \| Callable[..., object]	Requires a rule: HasRule("blog.change_post") or HasRule(is_author).

Task queue adapters (ninja_devx.contrib.tasks)

Task queues

Class	Arguments	Description
CeleryTaskQueue	DEFAULT_METHODS: tuple[str, ...] = ('delay',)	Celery: task.delay(*args, **kwargs).
DramatiqTaskQueue	DEFAULT_METHODS: tuple[str, ...] = ('send',)	Dramatiq: actor.send(*args, **kwargs).
RQTaskQueue	queue: object	RQ: queue.enqueue(function, *args, **kwargs).
TaskiqTaskQueue	DEFAULT_METHODS: tuple[str, ...] = ('kiq',)	Taskiq: task.kiq(*args, **kwargs).
TemporalTaskQueue	client: object	Temporal: client.start_workflow(workflow, *args, **kwargs).
FastStreamTaskQueue	broker: object, *, queue: str = ''	FastStream: publish {"task", "args", "kwargs"} through a broker.
FrameworkTaskQueue	*, queue: object \| None = None, client: object \| None = None, methods: Sequence[str] \| None = None	Defer work using a task framework's callable conventions.
DeferredTaskQueue	defer: Defer	TaskQueue over any defer(function, args, kwargs) callable.