API reference

Rendered docstrings for every public module. For focused tables of class attributes, decorator options, settings, commands and error codes, use the configuration reference instead; the guides link to the relevant page.

Controllers and operations

ninja_devx.routing.controller

Class-based controllers compiled into native Django Ninja routers.

Scope

Bases: StrEnum

How long a controller instance lives.

REQUEST class-attribute instance-attribute

REQUEST = 'request'

A new instance per request (like Django's class-based views).

SINGLETON class-attribute instance-attribute

SINGLETON = 'singleton'

One instance per as_router() call, created eagerly. Must be stateless.

ControllerOptions

Bases: TypedDict

Defaults for every operation of a controller.

Operation-level auth, throttle, tags, permissions, atomic and deprecated replace these; decorators and hooks are combined (the controller's wrap the operation's).

auth instance-attribute

auth: AuthSpec

Ninja authentication for every operation (django_auth, JWTAuth(), a list, or None).

throttle instance-attribute

throttle: ThrottleSpec

Ninja throttles for every operation (UserRateThrottle("100/min") or a list).

tags instance-attribute

tags: Sequence[str]

OpenAPI tags.

permissions instance-attribute

permissions: Sequence[AnyPermission]

Permissions checked before each operation (see Also for adding at operation level).

decorators instance-attribute

decorators: Sequence[ViewDecorator]

View decorators (paginate(...), conditional()); the first item is outermost.

hooks instance-attribute

hooks: Sequence[OperationHook | AsyncOperationHook]

Operation hooks wrapping each call (sync around or async around_async).

atomic instance-attribute

atomic: bool | Literal['durable']

Run sync operations in transaction.atomic(); "durable" must be the outermost.

database instance-attribute

database: str

Database alias for atomic.

errors instance-attribute

errors: ErrorMap

Exception rules for every operation (see ninja_devx.http.errors).

meta instance-attribute

meta: Sequence[object]

Typed metadata read with get_operation(request).meta(Kind).

plugins instance-attribute

plugins: Sequence[ControllerPlugin]

ControllerPlugin objects applied to every operation.

middleware instance-attribute

middleware: Sequence[Middleware]

Router middleware around every operation, including auth and throttling (RequestIDMiddleware(), DeprecationMiddleware(sunset=...)).

allow_mixed_path instance-attribute

allow_mixed_path: bool

Silence the warning for a path served by both sync and async operations.

deprecated instance-attribute

deprecated: bool

Mark every operation deprecated in OpenAPI.

document_errors instance-attribute

document_errors: bool

Document 401/403/404/422 responses in OpenAPI (default from settings).

by_alias instance-attribute

by_alias: bool

Serialize responses by field alias (Ninja).

exclude_unset instance-attribute

exclude_unset: bool

Leave unset fields out of responses (Ninja).

exclude_defaults instance-attribute

exclude_defaults: bool

Leave fields equal to their default out of responses (Ninja).

exclude_none instance-attribute

exclude_none: bool

Leave None fields out of responses (Ninja).

operation_id_prefix instance-attribute

operation_id_prefix: str

Prepended to every operation id (mount(prefix=...) sets it for versions).

url_name_prefix instance-attribute

url_name_prefix: str

Prepended (with _) to explicit url_name values.

BuiltRouter dataclass

What as_router() built a router from (used by system checks).

Controller

Base class for class-based Django Ninja controllers.

Declare operations with @get, @post... on instance methods, receive dependencies through __init__ and mount with::

api.add_router("/users", UserController.as_router(container=container))

Request data is never stored on self; operations receive request explicitly.

scope class-attribute

scope: Scope = Scope.REQUEST

Scope.REQUEST: a controller per request (inside the DI scope); Scope.SINGLETON: one per as_router() call.

mode class-attribute

mode: Literal['sync', 'async', 'auto'] = 'sync'

Which implementation to register when an operation has an async_variant; "auto" follows NINJA_DEVX["ASYNC_MODE"].

routes class-attribute

routes: Mapping[str, RouteOptions] = MappingProxyType({})

Overrides per operation name: {"bulk_create": {"path": "/batch"}, "destroy": {"enabled": False}, "list": {"summary": "Posts"}}. Unknown names fail at startup.

options class-attribute

options: ControllerOptions = ControllerOptions()

Merged along the MRO over NINJA_DEVX["DEFAULT_OPTIONS"], then as_router(**options).

as_router classmethod

as_router(
    *,
    container: ContainerLike | None = None,
    scope: Scope | None = None,
    **options: Unpack[ControllerOptions],
) -> Router

Build a new native Ninja Router exposing this controller's operations.

Every call returns an independent router, so the same controller can be mounted on several APIs or prefixes with different containers and scopes.

Parameters:

Name	Type	Description	Default
container	ContainerLike | None	Builds controllers and Inject[...] values: a Container, a dishka/svcs resolver, or any Resolver/RequestScopeProvider.	None
scope	Scope | None	Overrides the class's scope for this router.	None
options	Unpack[ControllerOptions]	ControllerOptions over the class's options.	{}

checks classmethod

checks(
    container: ContainerLike | None = None,
) -> list[CheckMessage]

Django system check messages for this controller (manage.py check).

container is the one it was mounted with. Override to add project rules; extend super().checks(container) to keep the built-in ones.

implementation classmethod

implementation(
    name: str, func: MethodFunction
) -> MethodFunction

The method registered for operation name: its async variant in async mode.

customize_operation classmethod

customize_operation(
    name: str, spec: OperationSpec
) -> OperationSpec

Adjust an operation before registration; name is the method name.

operation_bindings classmethod

operation_bindings(
    name: str, spec: OperationSpec
) -> Sequence[ParameterBinding]

Extra hidden parameters for an operation (nested resources use this).

documented_errors classmethod

documented_errors(
    name: str, spec: OperationSpec
) -> frozenset[int]

Extra error status codes to document for an operation.

before_operation

before_operation(
    request: HttpRequest, operation: OperationInfo
) -> object

Called after permissions and bindings, right before the method. May be async.

after_operation

after_operation(
    request: HttpRequest,
    operation: OperationInfo,
    result: object,
) -> object

Called with the method's result; the return value is sent. May be async.

authorize_replay

authorize_replay(
    request: HttpRequest,
    operation: OperationInfo,
    arguments: Mapping[str, object],
) -> object

Authorize every keyed attempt, including retries, after bindings and preflight.

Move authorization performed inside a custom handler here (or to bindings or before_operation). May be async. Model controllers also recheck the URL object.

resolve

resolve(request: HttpRequest, key: Callable[..., T]) -> T

Resolve key from this call's DI scope (e.g. a service chosen at runtime).

run_sync async

run_sync(
    function: Callable[P, R],
    /,
    *args: args,
    **kwargs: kwargs,
) -> R

Run sync (ORM) code from an async operation in one thread hop.

run_atomic async

run_atomic(
    function: Callable[P, R],
    /,
    *args: args,
    **kwargs: kwargs,
) -> R

Run sync code in transaction.atomic() from an async operation, in one hop.

check_object_permissions

check_object_permissions(
    request: HttpRequest, obj: object
) -> None

Enforce the current operation's object-level permissions on obj.

built_router

built_router(router: Router) -> BuiltRouter | None

The controller behind router when as_router() created it.

built_routers

built_routers() -> list[BuiltRouter]

Every live router as_router() created.

ninja_devx.routing.operations

Operation decorators for controller methods.

The decorators only attach metadata; nothing is registered until Controller.as_router() runs. Options mirror ninja.Router.api_operation one-to-one, plus permissions and decorators.

OperationMethod module-attribute

OperationMethod = Callable[Concatenate[C, RequestT, P], R]

A controller method: (self, request, *params) -> R.

OperationOptions

Bases: TypedDict

Keyword arguments accepted by every operation decorator.

auth instance-attribute

auth: AuthSpec

Ninja authentication for this operation; overrides the controller's.

throttle instance-attribute

throttle: ThrottleSpec

Ninja throttles for this operation.

response instance-attribute

response: ResponseSpec

Response schema, or {status: schema} (Ninja).

operation_id instance-attribute

operation_id: str | None

OpenAPI operation id (default <controller>_<method>).

summary instance-attribute

summary: str | None

OpenAPI summary.

description instance-attribute

description: str | None

OpenAPI description (default: the method docstring).

tags instance-attribute

tags: list[str] | None

OpenAPI tags.

deprecated instance-attribute

deprecated: bool | None

Mark the operation deprecated in OpenAPI.

by_alias instance-attribute

by_alias: bool | None

Serialize the response by field alias (Ninja).

exclude_unset instance-attribute

exclude_unset: bool | None

Leave unset fields out of the response (Ninja).

exclude_defaults instance-attribute

exclude_defaults: bool | None

Leave fields equal to their default out of the response (Ninja).

exclude_none instance-attribute

exclude_none: bool | None

Leave None fields out of the response (Ninja).

url_name instance-attribute

url_name: str | None

Django URL name for reverse().

include_in_schema instance-attribute

include_in_schema: bool

Show the operation in OpenAPI.

openapi_extra instance-attribute

openapi_extra: dict[str, JSONValue] | None

Merged into the operation's OpenAPI object.

permissions instance-attribute

permissions: Sequence[AnyPermission]

Replaces the controller's permissions; Also(...) adds to them instead.

decorators instance-attribute

decorators: Sequence[ViewDecorator]

View decorators applied inside the controller's ones; the first item is outermost.

hooks instance-attribute

hooks: Sequence[OperationHook | AsyncOperationHook]

Operation hooks run inside the controller's hooks.

atomic instance-attribute

atomic: bool | Literal['durable']

Run the operation in transaction.atomic() ("durable": must be the outermost).

database instance-attribute

database: str

Database alias for atomic.

errors instance-attribute

errors: ErrorMap

Exception rules for this operation, over the controller's.

raises instance-attribute

raises: Sequence[type[BaseException]]

Exceptions this operation may raise: documented in OpenAPI, checked against the rules.

meta instance-attribute

meta: Sequence[object]

Typed metadata read by permissions and hooks via get_operation(request).meta(Kind).

document_errors instance-attribute

document_errors: bool

Document 401/403/404/422 responses in OpenAPI (default from settings).

RouteOptions

Bases: OperationOptions

Per-operation overrides in Controller.routes (paths and options are not hard-coded).

path instance-attribute

path: str

Replaces the declared path, e.g. {"restore": {"path": "/{pk}/undelete"}}.

enabled instance-attribute

enabled: bool

False removes the operation from the router.

async_variant

async_variant(
    sync_method: Callable[..., object],
) -> Callable[[AsyncF], AsyncF]

Declare the async implementation of an operation, used when the controller runs async.

Reusable bases (like the CRUD mixins) define both and let mode pick::

@get("/")
def list(self, request): ...

@async_variant(list)
async def alist(self, request): ...

Parameters:

Name	Type	Description	Default
sync_method	Callable[..., object]	The sync operation this async method implements.	required

api_operation

api_operation(
    methods: str | Sequence[str],
    path: str = "/",
    **options: Unpack[OperationOptions],
) -> Callable[
    [OperationMethod[C, RequestT, P, R]],
    OperationMethod[C, RequestT, P, R],
]

Mark a controller method as an operation for the given HTTP methods.

Can be stacked to expose the same method under several paths.

Parameters:

Name	Type	Description	Default
methods	str | Sequence[str]	HTTP methods, e.g. ["GET", "HEAD"].	required
path	str	Path relative to the router; {name} segments become parameters.	'/'
options	Unpack[OperationOptions]	OperationOptions: Ninja's operation keywords plus ninja-devx's.	{}

query

query(
    path: str = "/", **options: Unpack[OperationOptions]
) -> Callable[
    [OperationMethod[C, RequestT, P, R]],
    OperationMethod[C, RequestT, P, R],
]

An HTTP QUERY operation: safe and idempotent like GET, with a request body.

Permissions treat it as a read (view). The OpenAPI document lists it under the query key that OpenAPI 3.2 defines; tools that only know 3.1 skip it.

TODO: Django has no QUERY support of its own yet. CsrfViewMiddleware treats it as unsafe (session-authenticated calls need the CSRF token) and django.test.Client has no query() (use client.request("QUERY", ...)). Revisit when Django adds it.

ninja_devx.dependencies.injection

Inject dependencies into operation parameters: service: Inject[OrderService].

::

@post("/", response={201: OrderOut})
async def place(self, request: HttpRequest, payload: OrderIn,
                orders: Inject[OrderService]) -> Status[Order]: ...

@get("/whoami")
def whoami(
    self, request: HttpRequest, ip: Annotated[str, Resolve(client_ip)]
) -> dict[str, str]: ...

Injected parameters never appear in OpenAPI. Inject resolves from the controller's container inside the request scope (checked when the router is built); Resolve calls a function with the request.

Inject module-attribute

Inject: TypeAlias = Annotated[T, _InjectMarker()]

A parameter resolved from the controller's container (not part of the API).

Resolve

Bases: BindingMarker, Generic[T]

Annotated[T, Resolve(fn)]: the value of fn(request) (sync or async fn).

resolve

resolve(request: HttpRequest, key: Callable[..., T]) -> T

Resolve key from the current operation's DI scope (for code without parameters).

injected

injected(key: object) -> object

The runtime form of Inject[key] for dynamically built signatures.

ninja_devx.dependencies.instances

How controller instances (and their DI scope) are provided per call.

Scope

Bases: StrEnum

How long a controller instance lives.

REQUEST class-attribute instance-attribute

REQUEST = 'request'

A new instance per request (like Django's class-based views).

SINGLETON class-attribute instance-attribute

SINGLETON = 'singleton'

One instance per as_router() call, created eagerly. Must be stateless.

Invocation

One call of an operation: the controller, the request and its DI scope.

InstanceProvider dataclass

Creates (or reuses) the controller for one call, within a DI scope if available.

factory

factory() -> Callable[[], Controller] | None

A plain callable when no scope has to be entered (the fast path).

get_invocation

get_invocation(request: HttpRequest) -> Invocation | None

The current invocation (controller and DI scope) of request.

ninja_devx.routing.use_cases

Bind an operation to a use case (command handler) without writing a method body.

::

class CreatePost:
    def __init__(self, posts: PostRepository) -> None: ...
    def __call__(self, command: CreatePostCommand) -> Post: ...

class PostController(Controller):
    create = use_case(
        post("/", response={201: PostOut}),
        CreatePost,
        command=PostIn.to_command,   # Callable[[PostIn], CreatePostCommand]
        status=201,
    )

The use case is resolved from the controller's container per request; the payload type is the command mapper's parameter annotation.

QueryHandler

Bases: Protocol[CommandT_contra, ResultT_co]

A read handler: __call__(query) sync or async def __call__.

use_case

use_case(
    decorator: Callable[[Method], Method],
    handler: Callable[
        ...,
        UseCase[CommandT, ResultT]
        | UseCase[CommandT, Awaitable[ResultT]],
    ],
    *,
    command: Callable[[PayloadT], CommandT],
    status: int | None = None,
) -> Method

An operation method: map the payload with command and call the resolved handler.

An async handler (async def __call__) makes the operation async.

Parameters:

Name	Type	Description	Default
decorator	Callable[[Method], Method]	The operation decorator, e.g. post("/", response={201: OrderOut}).	required
handler	Callable[..., UseCase[CommandT, ResultT] | UseCase[CommandT, Awaitable[ResultT]]]	A class with __call__(command) (sync or async), resolved from the container.	required
command	Callable[[PayloadT], CommandT]	Maps the validated payload to the command; its parameter type is the request body.	required
status	int | None	Status code of the response (default: the operation's first response).	None

use_query

use_query(
    decorator: Callable[[Method], Method],
    handler: Callable[
        ...,
        QueryHandler[CommandT, ResultT]
        | QueryHandler[CommandT, Awaitable[ResultT]],
    ],
    *,
    query: Callable[[PayloadT], CommandT],
    status: int | None = None,
) -> Method

A read operation: map the query parameters and call the resolved handler.

The read counterpart of :func:use_case. Use a FilterSchema (or any schema) as the handler's payload; for GET operations its fields become query parameters.

Parameters:

Name	Type	Description	Default
decorator	Callable[[Method], Method]	The operation decorator, e.g. get("/stats", response=StatsOut).	required
handler	Callable[..., QueryHandler[CommandT, ResultT] | QueryHandler[CommandT, Awaitable[ResultT]]]	A class with __call__(query) (sync or async), resolved from the container.	required
query	Callable[[PayloadT], CommandT]	Maps the validated payload to the query; its parameter type is the query schema.	required
status	int | None	Status code of the response (default: the operation's first response).	None

ninja_devx.routing.plugins

Plugins extend every controller they are given to, without touching the package.

::

class TenantHeader:
    def on_operation(
        self, controller: type[Controller], name: str, spec: OperationSpec
    ) -> OperationSpec:
        return spec.with_options(openapi_extra={"parameters": [...]})

    def bindings(
        self, controller: type[Controller], name: str, spec: OperationSpec
    ) -> Sequence[ParameterBinding]:
        return ()

options = ControllerOptions(plugins=[TenantHeader()])   # or NINJA_DEVX["PLUGINS"]

Plugins are listed explicitly; nothing is discovered through entry points.

ControllerPlugin

Bases: Protocol

on_operation

on_operation(
    controller: type[Controller],
    name: str,
    spec: OperationSpec,
) -> OperationSpec

Adjust an operation before it is registered (after customize_operation).

bindings

bindings(
    controller: type[Controller],
    name: str,
    spec: OperationSpec,
) -> Sequence[ParameterBinding]

Extra hidden parameters for the operation.

ninja_devx.routing.bindings

Parameters whose value is computed per call instead of passed by Ninja.

A binding exposes its own parameters to Ninja (e.g. a pk path parameter), or none (injected services), and turns them into the value the method receives.

ParameterBinding dataclass

How one method parameter (or a hidden one, when name is None) is produced.

resolve pops its exposed parameters from the arguments and returns the value; aresolve is used by async operations when given. check validates the binding against the container when the router is built.

BindingMarker

Bases: LazyAnnotation

Annotated metadata that turns a parameter into a ParameterBinding.

Permissions, users and errors

ninja_devx.security.permissions

Typed, composable permissions evaluated after Django Ninja has validated the request.

Authentication stays Ninja's job (auth=); permissions decide what an authenticated (or anonymous) caller may do::

@get("/{pk}", auth=django_auth, permissions=[IsAuthenticated() & IsStaff()])

Every permission works in sync and async operations: async operations call ahas_permission/ahas_object_permission, which default to the sync checks and are overridden by the built-ins to load session users without blocking.

Permissions are immutable and shared between requests, so they must not store state.

PermissionResult module-attribute

PermissionResult: TypeAlias = bool | Awaitable[bool]

Sync permissions return bool; a coroutine has_permission makes it async-only.

AnyPermission module-attribute

AnyPermission: TypeAlias = 'BasePermission[Never]'

Any permission, whatever object type it checks (permissions are contravariant).

BasePermission

Bases: Generic[ObjT_contra]

Allow everything by default; override the checks you need.

has_permission runs before the operation. has_object_permission runs when the operation loads an object (get_object, Instance[...]) or calls check_object_permissions. Override the a-prefixed versions for native async checks. Combine with &, | and ~.

message class-attribute instance-attribute

message: str = gettext_noop(
    "You do not have permission to perform this action."
)

detail of the denial response, translated with gettext when sent (so your own messages can come from your project's catalog).

status_code class-attribute instance-attribute

status_code: int = 403

Status of the denial response (403, or 401 for authentication).

combinator class-attribute

combinator: Literal['all', 'any', 'not'] | None = None

Set by AllOf/AnyOf/Not, which evaluate operands instead of checks.

Also

Bases: tuple['BasePermission[Never]', ...]

Operation permissions added to the controller's instead of replacing them.

@get("/{pk}", permissions=Also(IsOwner()))

AllOf dataclass

Bases: BasePermission[ObjT_contra]

Allowed when every permission allows; reports the first denial.

AnyOf dataclass

Bases: BasePermission[ObjT_contra]

Allowed when at least one permission allows; reports the last denial.

Not dataclass

Bases: BasePermission[ObjT_contra]

Inverts a permission. Uses its own message/status_code when denying.

AllowAny

Bases: BasePermission[object]

Allows every request.

DenyAll

Bases: BasePermission[object]

Denies every request (403).

IsAuthenticated

Bases: BasePermission[object]

Requires Ninja authentication (request.auth) or an authenticated request.user.

IsAuthenticatedOrReadOnly

Bases: IsAuthenticated

Safe methods (GET, HEAD, OPTIONS) for everyone; other methods need authentication.

IsReadOnly

Bases: BasePermission[object]

Allows safe methods only; combine it: [IsAuthenticated(), IsReadOnly() | IsStaff()].

IsStaff

Bases: BasePermission[object]

The user has is_staff.

IsSuperuser

Bases: BasePermission[object]

The user has is_superuser.

HasDjangoPermission dataclass

Bases: BasePermission[object]

Requires Django model permissions, e.g. HasDjangoPermission("blog.change_post").

perms instance-attribute

perms: tuple[str, ...]

Django permission names ("app.change_model"), all required.

DjangoModelPermissions dataclass

Bases: BasePermission[object]

Maps the HTTP method to the model's view/add/change/delete permissions.

The model is the controller's (ModelController) unless given explicitly. Unknown HTTP methods are denied.

IsOwner dataclass

Bases: BasePermission[object]

Object-level: obj.<field> must be the current user.

field may follow relations: IsOwner("author"), IsOwner("customer__user"). ModelController selects the related objects so no extra query runs.

PolicyPermission dataclass

Bases: BasePermission[ObjT], Generic[SubjectT, ObjT]

A Policy enforced on objects loaded by an operation (see as_permission).

as_permission

as_permission(
    policy: Policy[SubjectT, ObjT],
    subject: type[SubjectT]
    | Callable[[HttpRequest], SubjectT],
    *,
    asubject: Callable[[HttpRequest], Awaitable[SubjectT]]
    | None = None,
) -> PolicyPermission[SubjectT, ObjT]

Use a service-layer Policy as an object permission.

subject is the user class (the authenticated user must be one, else 401) or a function of the request::

permissions=[as_permission(CanEdit(), User)]
permissions=[as_permission(CanEdit(), membership_of, asubject=amembership_of)]

Parameters:

Name	Type	Description	Default
policy	Policy[SubjectT, ObjT]	An object with allows(subject, obj) -> bool.	required
subject	type[SubjectT] | Callable[[HttpRequest], SubjectT]	The user class (the authenticated user must be one, else 401), or a function of the request.	required
asubject	Callable[[HttpRequest], Awaitable[SubjectT]] | None	Async version of subject for async operations.	None

requires_async

requires_async(permission: AnyPermission) -> bool

Whether permission can only be evaluated asynchronously (coroutine checks).

bind_permissions

bind_permissions(
    request: HttpRequest,
    permissions: Sequence[AnyPermission],
) -> None

Remember the operation's permissions for later object-level checks.

check_object_permissions

check_object_permissions(
    request: HttpRequest, obj: object
) -> None

Raise HttpError unless the current operation's permissions allow obj.

ninja_devx.security.object_permissions

Object-level permissions: per-row grants, checked on objects and applied to lists.

The model follows django-guardian and DRF's DjangoObjectPermissions: permissions are Django's app_label.codename strings (blog.change_post), granted to users or groups on single objects. A backend stores and checks them:

==================== ================================================================== GrantsBackend ninja_devx.contrib.grants: no extra dependency GuardianBackend django-guardian (pip install ninja-devx[guardian]) DjangoBackend any authentication backend's user.has_perm(perm, obj); checks only, lists cannot be filtered ==================== ==================================================================

NINJA_DEVX["OBJECT_PERMISSION_BACKEND"] picks one (an instance or import path). By default: grants when ninja_devx.contrib.grants is installed, else guardian when installed, else DjangoBackend.

::

class DocumentController(CRUDController[Document, DocumentOut, DocumentIn]):
    object_permissions = ObjectPermissions()      # 404 without view, 403 without change

assign_perm("docs.change_document", bob, document)
get_objects_for_user(bob, "docs.view_document", Document.objects.all())

Grant dataclass

Permissions a user or a group holds on one object.

ObjectPermissionBackend

Bases: Protocol

has_perm

has_perm(user: object, perm: str, obj: Model) -> bool

Whether user holds perm on obj.

filter_queryset

filter_queryset(
    user: object,
    perms: Sequence[str],
    queryset: QuerySet[ModelT],
) -> QuerySet[ModelT]

The objects of queryset on which user holds every one of perms.

DjangoBackend

user.has_perm(perm, obj) through AUTHENTICATION_BACKENDS; no list filtering.

GuardianBackend

django-guardian's ObjectPermissionChecker and shortcuts.

ObjectPermissions dataclass

Bases: BasePermission[Model]

Per-object Django permissions by HTTP method, like DRF's DjangoObjectPermissions.

• Detail operations need the object permissions of the method (view for GET, change for PUT/PATCH, delete for DELETE).
• A caller who cannot even view the object gets 404, so existence doesn't leak.
• With filter_lists (used by ModelController.object_permissions), queries only return objects the caller can view.
• model_permissions=True also accepts a model-wide permission (user.has_perm("blog.change_post")) for users with global rights.

perms_map class-attribute instance-attribute

perms_map: Mapping[str, Sequence[str]] = field(
    default_factory=lambda: dict(DEFAULT_PERMS_MAP)
)

HTTP method → permission templates (%(app_label)s, %(model_name)s).

model_permissions class-attribute instance-attribute

model_permissions: bool = False

Accept model-wide Django permissions in addition to object grants.

filter_lists class-attribute instance-attribute

filter_lists: bool = True

Restrict querysets to objects the caller can view (needs a filtering backend).

hide_forbidden class-attribute instance-attribute

hide_forbidden: bool = True

Answer 404 instead of 403 when the caller cannot view the object.

filter

filter(
    request: HttpRequest, queryset: QuerySet[ModelT]
) -> QuerySet[ModelT]

queryset restricted to objects the caller can view.

register_object_permission_backend

register_object_permission_backend(
    name: str, factory: BackendFactory
) -> None

Register a backend so get_backend can select it by name.

Apps register in AppConfig.ready (ninja_devx.contrib.grants registers "grants"); core never imports contrib.

registered_object_permission_backends

registered_object_permission_backends() -> tuple[str, ...]

Names of the currently registered backends.

assign_perm

assign_perm(perm: str, holder: object, obj: Model) -> None

Grant perm ("app_label.codename") to a user or group on obj.

Parameters:

Name	Type	Description	Default
perm	str	Full permission name, "app_label.codename".	required
holder	object	A user or a Group.	required
obj	Model	The model instance.	required

remove_perm

remove_perm(perm: str, holder: object, obj: Model) -> None

Revoke perm from a user or group on obj.

Parameters:

Name	Type	Description	Default
perm	str	Full permission name, "app_label.codename".	required
holder	object	A user or a Group.	required
obj	Model	The model instance.	required

grants_for

grants_for(obj: Model) -> list[Grant]

Every user and group holding permissions on obj.

Parameters:

Name	Type	Description	Default
obj	Model	The model instance.	required

get_perms

get_perms(
    user: object, obj: Model, perms: Sequence[str]
) -> list[str]

The subset of perms that user holds on obj.

Parameters:

Name	Type	Description	Default
user	object	The user (anonymous users hold nothing).	required
obj	Model	The model instance.	required
perms	Sequence[str]	Full permission names to test.	required

get_objects_for_user

get_objects_for_user(
    user: object,
    perms: str | Sequence[str],
    queryset: QuerySet[ModelT],
) -> QuerySet[ModelT]

Objects of queryset on which user holds every one of perms.

Parameters:

Name	Type	Description	Default
user	object	The user.	required
perms	str | Sequence[str]	One or several full permission names, all required.	required
queryset	QuerySet[ModelT]	The objects to filter.	required

ninja_devx.security.auth

Typed access to the authenticated user, and building a RequestContext.

AuthedRequest

Bases: HttpRequest, Generic[UserT]

An HttpRequest whose auth (set by Ninja authentication) is UserT.

Use it to annotate request on operations that require authentication::

@get("/me")
def me(self, request: AuthedRequest[User]) -> User:
    return request.auth

request_user

request_user(request: HttpRequest) -> object | None

The authenticated user: request.auth when it is a user, else request.user.

An unevaluated lazy request.user is not loaded inside an event loop (that would query the session synchronously); use arequest_user there.

arequest_user async

arequest_user(request: HttpRequest) -> object | None

request_user for async code: loads the session user with request.auser().

current_user

current_user(
    request: HttpRequest, user_type: type[UserT]
) -> UserT

The authenticated user_type instance; raises AuthenticationError (401) otherwise.

authenticated_user

authenticated_user(
    user_type: type[UserT],
) -> Callable[[HttpRequest], UserT]

A scoped factory: container.scoped(User, authenticated_user(User)).

Parameters:

Name	Type	Description	Default
user_type	type[UserT]	The user class; the authenticated user must be one, else 401.	required

aauthenticated_user

aauthenticated_user(
    user_type: type[UserT],
) -> Callable[[HttpRequest], Awaitable[UserT]]

authenticated_user for async containers and as_permission(asubject=...).

request_context

request_context(
    user_type: type[UserT], tenant: None = None
) -> Callable[[HttpRequest], RequestContext[UserT, None]]

request_context(
    user_type: type[UserT],
    tenant: Callable[[HttpRequest, UserT], TenantT],
) -> Callable[
    [HttpRequest], RequestContext[UserT, TenantT]
]

request_context(
    user_type: type[UserT],
    tenant: Callable[[HttpRequest, UserT], TenantT]
    | None = None,
) -> (
    Callable[[HttpRequest], RequestContext[UserT, TenantT]]
    | Callable[[HttpRequest], RequestContext[UserT, None]]
)

A scoped factory building RequestContext from the request.

::

container.scoped(RequestContext[User, None], request_context(User))
container.scoped(RequestContext[User, Org], request_context(User, tenant=org_of))

The request id comes from X-Request-ID and the trace id from traceparent when present. In async operations the async container uses arequest_context.

Parameters:

Name	Type	Description	Default
user_type	type[UserT]	The user class; the authenticated user must be one, else 401.	required
tenant	Callable[[HttpRequest, UserT], TenantT] | None	Returns the tenant from (request, user); None for single-tenant apps.	None

arequest_context

arequest_context(
    user_type: type[UserT], tenant: None = None
) -> Callable[
    [HttpRequest], Awaitable[RequestContext[UserT, None]]
]

arequest_context(
    user_type: type[UserT],
    tenant: Callable[[HttpRequest, UserT], TenantT],
) -> Callable[
    [HttpRequest], Awaitable[RequestContext[UserT, TenantT]]
]

arequest_context(
    user_type: type[UserT],
    tenant: Callable[[HttpRequest, UserT], TenantT]
    | None = None,
) -> (
    Callable[
        [HttpRequest],
        Awaitable[RequestContext[UserT, TenantT]],
    ]
    | Callable[
        [HttpRequest],
        Awaitable[RequestContext[UserT, None]],
    ]
)

request_context for async operations (loads session users without blocking).

ninja_devx.http.errors

Map exceptions (domain errors, Django errors, your own) to HTTP responses.

Rules layer like options: operation errors → controller errors → NINJA_DEVX["ERRORS"] → the defaults. Exceptions implementing ninja_devx.layers.HttpMappable (every DomainError) map themselves.

At the API level ErrorMap.install(api) registers the same rules with Ninja's own api.add_exception_handler; mount(api, ..., errors=...) does it for you.

ErrorRule dataclass

Bases: Generic[E]

exception instance-attribute

exception: type[E]

Exception class (subclasses match too).

status instance-attribute

status: int

HTTP status.

code instance-attribute

code: str

Machine-readable code.

body class-attribute instance-attribute

body: Callable[[E], Mapping[str, JSONValue]] | None = None

Builds the body from the exception (default {detail, code}).

ErrorMap

An immutable set of exception → response rules. Later rules win.

django_defaults classmethod

django_defaults() -> ErrorMap

Django's ValidationError → 422, ObjectDoesNotExist → 404, PermissionDenied → 403.

map

map(
    exception: type[E],
    status: int,
    *,
    code: str | None = None,
    body: Callable[[E], Mapping[str, JSONValue]]
    | None = None,
) -> ErrorMap

A new map where exception (and subclasses) produce status.

Parameters:

Name	Type	Description	Default
exception	type[E]	Exception class; subclasses match too.	required
status	int	HTTP status of the response.	required
code	str | None	Machine-readable code (default: the class name in snake_case).	None
body	Callable[[E], Mapping[str, JSONValue]] | None	Builds the body from the exception (default {detail, code} or error_body()).	None

rule_for

rule_for(exception: type[BaseException]) -> AnyRule | None

The rule for exception: closest class in its MRO, later rules first.

install

install(api: NinjaAPI) -> None

Register the rules with Ninja's api.add_exception_handler.

mask_validation_input

mask_validation_input(
    errors: Iterable[Mapping[str, JSONValue]],
    schema: type[object] | None = None,
) -> list[dict[str, JSONValue]]

Mask Sensitive field values a validation error body would otherwise echo back.

Ninja's own request validation already drops pydantic's input key from 422 bodies; this matters for a raw pydantic.ValidationError you map yourself and whose .errors() keeps it::

ErrorMap().map(
    PydanticValidationError,
    422,
    code="validation_failed",
    body=lambda exc: {"detail": mask_validation_input(exc.errors(), MySchema)},
)

Parameters:

Name	Type	Description	Default
errors	Iterable[Mapping[str, JSONValue]]	Error items shaped like pydantic's/Ninja's (a loc, optionally an input).	required
schema	type[object] | None	The schema errors were raised against; without one nothing is masked.	None

ninja_devx.exceptions

NinjaDevXError

Bases: Exception

Base class for all ninja-devx errors.

ControllerConfigError

Bases: NinjaDevXError

A controller or one of its operations is declared incorrectly.

DependencyResolutionError

Bases: NinjaDevXError

A dependency could not be resolved by the container.

CircularDependencyError

Bases: DependencyResolutionError

The dependency graph contains a cycle.

AsyncLazyAccessError

Bases: NinjaDevXError

Sync-only Django code (usually a lazy relation) ran inside an async operation.

BlockingCallWarning

Bases: RuntimeWarning

Sync code blocked an async operation's event loop (see NINJA_DEVX["WARN_BLOCKING_MS"]).

AsyncDatabaseTestWarning

Bases: UserWarning

An async test uses the database without django_db(transaction=True).

sync_to_async runs ORM calls on another thread with its own connection, outside the test's transaction: rows leak between tests or are invisible to the view.

MixedPathWarning

Bases: UserWarning

One path is served by both sync and async operations (costs a thread hop per call).

NinjaDevXDeprecationWarning

Bases: DeprecationWarning

A ninja-devx API scheduled for removal; the message names the replacement.

filterwarnings = ["error::ninja_devx.NinjaDevXDeprecationWarning"] turns every use into a test failure before upgrading.

Cross-cutting

ninja_devx.routing.hooks

Operation metadata and hooks around every operation call.

OperationInfo dataclass

Static description of an operation, built once per registration.

controller instance-attribute

controller: type[Controller]

The controller class.

method_name instance-attribute

method_name: str

The Python method implementing the operation.

operation_id instance-attribute

operation_id: str

The OpenAPI operation id.

http_methods instance-attribute

http_methods: tuple[str, ...]

HTTP methods of the operation.

path instance-attribute

path: str

The path as declared on the router.

is_async instance-attribute

is_async: bool

Whether the registered implementation is async.

metadata class-attribute instance-attribute

metadata: tuple[object, ...] = ()

Typed metadata from meta= (controller and operation), for permissions and hooks.

database class-attribute instance-attribute

database: str | None = None

Explicit operation/controller database alias, if configured.

meta

meta(kind: type[MetaT]) -> MetaT | None

The closest metadata item of kind: operation.meta(RequiresScope).

OperationHook

Bases: Protocol

Wraps each call of an operation, including permission checks.

Exceptions (denials, 404s, errors) propagate through the context manager, so hooks can observe them. Hooks are shared between requests and must not store per-request state on self.

AsyncOperationHook

Bases: Protocol

A hook with an async context manager, used by async operations.

A hook may implement both around and around_async; async operations prefer around_async. Sync-only hooks also run in async operations and are assumed not to block; set blocking = True on them to run their enter/exit in a thread.

LoggingHook dataclass

Logs one structured record per operation call with its duration and outcome.

logger class-attribute instance-attribute

logger: Logger = field(
    default_factory=lambda: logging.getLogger("ninja_devx")
)

Logger receiving one record per call.

level class-attribute instance-attribute

level: int = logging.INFO

Log level of the record.

get_operation

get_operation(request: HttpRequest) -> OperationInfo | None

The operation handling request, available to permissions, services and hooks.

ninja_devx.dependencies.container

Dependency registration and container ownership.

Container

Constructor-injection container with singleton, scoped and transient lifetimes.

• Unregistered concrete classes are auto-wired as transient.
• Abstract classes, protocols and builtins must be registered.
• Factories may be generators (cleanup after yield), coroutines or async generators; async ones need aresolve/async with container.scope().
• scope() opens a scope anywhere (tasks, commands, tests); inside request_scope(request) the HttpRequest itself is injectable.

Keys are typed as Callable[..., T] rather than type[T] so abstract classes and protocols are accepted by mypy.

singleton

singleton(
    key: Factory[T], factory: Factory[T] | None = None
) -> None

One instance per container, built from factory (defaults to key).

Parameters:

Name	Type	Description	Default
key	Factory[T]	What is requested (a class, abstract class, protocol or generic alias).	required
factory	Factory[T] | None	Builds it: a class, function, generator or async variant (default: key itself).	None

scoped

scoped(
    key: Factory[T], factory: Factory[T] | None = None
) -> None

One instance per scope; unavailable outside of one.

Parameters:

Name	Type	Description	Default
key	Factory[T]	What is requested.	required
factory	Factory[T] | None	Builds it once per scope (request, or container.scope()); generators clean up when the scope closes.	None

transient

transient(
    key: Factory[T], factory: Factory[T] | None = None
) -> None

A new instance on every resolution.

Parameters:

Name	Type	Description	Default
key	Factory[T]	What is requested.	required
factory	Factory[T] | None	Builds a new value every time it is resolved.	None

register

register(
    key: Factory[T],
    factory: Factory[T] | None = None,
    *,
    lifetime: Lifetime,
) -> None

Parameters:

Name	Type	Description	Default
key	Factory[T]	What is requested.	required
factory	Factory[T] | None	Builds it (default: key itself).	None
lifetime	Lifetime	Lifetime.SINGLETON, SCOPED or TRANSIENT.	required

instance

instance(key: Factory[object], value: object) -> None

Register an already constructed object.

value is checked against key when the key is a (runtime-checkable) class: type checkers cannot infer it from abstract classes and protocols.

Parameters:

Name	Type	Description	Default
key	Factory[object]	What is requested.	required
value	object	The object returned; checked against class keys at registration.	required

override

override(
    key: Factory[object], value: object
) -> Generator[None]

Temporarily replace a dependency, e.g. with a fake in tests.

Singletons built earlier keep the dependency they were built with.

Parameters:

Name	Type	Description	Default
key	Factory[object]	The dependency to replace.	required
value	object	The replacement, until the with block ends.	required

resolve

resolve(key: Factory[T]) -> T

Resolve outside of a scope (scoped dependencies are rejected).

Parameters:

Name	Type	Description	Default
key	Factory[T]	What to build (sync factories only).	required

aresolve async

aresolve(key: Factory[T]) -> T

Parameters:

Name	Type	Description	Default
key	Factory[T]	What to build (sync and async factories).	required

scope

scope(
    values: Mapping[object, object] | None = None,
) -> Scope

A scope for scoped services, closed (with cleanups) when the block exits.

values are available for injection by key, e.g. container.scope({RequestContext[User, None]: context}).

Parameters:

Name	Type	Description	Default
values	Mapping[object, object] | None	Values available to scoped factories, e.g. {RequestContext[User, None]: context}.	None

close

close() -> None

Run the cleanup of generator singletons.

check

check(
    key: Factory[object], /, *, asynchronous: bool = False
) -> None

Validate the graph of key without building anything.

Raises DependencyResolutionError for keys that cannot be resolved, singletons depending on scoped services, and async factories needed by a sync resolution.

Parameters:

Name	Type	Description	Default
key	Factory[object]	What will be requested.	required
asynchronous	bool	Whether async operations will request it (allows async factories).	False

plan

plan(factory: Factory[object]) -> tuple[Dependency, ...]

Inspect a factory's parameters once and cache the result.

ninja_devx.routing.mounting

Mount many controllers at once, e.g. per API version.

Mount dataclass

Per-route overrides for mount.

controller instance-attribute

controller: type[Controller]

The controller class.

container class-attribute instance-attribute

container: ContainerLike | None = None

Container for this entry (default: the mount() container).

scope class-attribute instance-attribute

scope: Scope | None = None

Scope for this entry (default: the mount() scope).

options class-attribute instance-attribute

options: ControllerOptions = field(
    default_factory=lambda: ControllerOptions()
)

ControllerOptions for this entry, over the mount() options.

mount

mount(
    target: NinjaAPI | Router,
    routes: Mapping[str, type[Controller] | Mount],
    *,
    prefix: str = "",
    container: ContainerLike | None = None,
    scope: Scope | None = None,
    **options: Unpack[ControllerOptions],
) -> dict[str, Router]

Build and add a router per route; shared container, scope and options.

::

mount(api, {"/articles": ArticleController, "/authors": AuthorController}, container=c)
mount(api, V1_ROUTES, prefix="/v1", deprecated=True)
mount(api, V2_ROUTES, prefix="/v2")

With a prefix, operation ids and URL names get a matching prefix (v1_) so OpenAPI ids and URL reversing stay unique when the same controllers are mounted twice.

Parameters:

Name	Type	Description	Default
target	NinjaAPI | Router	The NinjaAPI or Router to mount on.	required
routes	Mapping[str, type[Controller] | Mount]	{prefix: Controller} or {prefix: Mount(Controller, ...)}.	required
prefix	str	Prepended to every route; also prefixes operation ids and URL names (versioning).	''
container	ContainerLike | None	Shared container for every entry.	None
scope	Scope | None	Shared scope for every entry.	None
options	Unpack[ControllerOptions]	ControllerOptions for every entry; errors= is also installed on a NinjaAPI.	{}

ninja_devx.idempotency

Authenticated, database-backed replay of completed controller responses.

idempotent

idempotent(
    *,
    header: str = "Idempotency-Key",
    ttl: int | None = None,
    database: str | None = None,
    required: bool = False,
    scope: Callable[[HttpRequest], str] | None = None,
) -> ViewDecorator

Replay completed responses after authentication, permissions and bindings.

Install ninja_devx and run its migrations. Unfinished requests remain claimed until explicitly reconciled; they never expire or repeat automatically. Completed responses (including errors) are replayed for ttl seconds. Custom authorization belongs in bindings, before_operation or authorize_replay, which run on retries.

Parameters:

Name	Type	Description	Default
header	str	Request header carrying a key of at most 255 characters.	'Idempotency-Key'
ttl	int | None	Completed response retention; defaults to IDEMPOTENCY_TTL.	None
database	str | None	Durable database alias; defaults to IDEMPOTENCY_DATABASE.	None
required	bool	Reject a missing key with HTTP 400.	False
scope	Callable[[HttpRequest], str] | None	Stable custom principal and tenant identity, if not model/scalar values.	None

ninja_devx.configuration.settings

Project-wide defaults read from settings.NINJA_DEVX.

::

NINJA_DEVX = {
    "DEFAULT_OPTIONS": ControllerOptions(auth=JWTAuth()),
    "PAGINATION_CLASS": "ninja.pagination.PageNumberPagination",
    "BULK_LIMIT": 500,
}

A class attribute set on a user controller always wins over these settings.

NinjaDevXSettings

Bases: TypedDict

DEFAULT_OPTIONS instance-attribute

DEFAULT_OPTIONS: ControllerOptions

Options applied below every controller's own options.

DOCUMENT_ERRORS instance-attribute

DOCUMENT_ERRORS: bool

Add 401/403/404/422 responses to OpenAPI (default True).

VALIDATE_MODEL instance-attribute

VALIDATE_MODEL: bool

Run Model.full_clean() on CRUD writes; errors are 422.

REFRESH_AFTER_WRITE instance-attribute

REFRESH_AFTER_WRITE: bool

Reload objects through scoped_queryset() after writes.

OPTIMIZE_QUERIES instance-attribute

OPTIMIZE_QUERIES: bool | Literal['only']

Derive select_related/prefetch_related from output schemas (default True).

PAGINATION_CLASS instance-attribute

PAGINATION_CLASS: type[PaginationBase] | str | None

Default pagination class for CRUD lists (a class or import path).

BULK_LIMIT instance-attribute

BULK_LIMIT: int

Maximum objects per bulk request.

IDEMPOTENCY_TTL instance-attribute

IDEMPOTENCY_TTL: int

Seconds a stored idempotent() response is replayed.

IDEMPOTENCY_DATABASE instance-attribute

IDEMPOTENCY_DATABASE: str

Database alias for durable idempotency records; must use autocommit.

ERRORS instance-attribute

ERRORS: ErrorMap | str

Project error rules (an ErrorMap or its import path) added to the defaults.

ERROR_FORMAT instance-attribute

ERROR_FORMAT: Literal['ninja', 'problem+json']

Error body format: "ninja" ({detail, code}) or "problem+json" (RFC 9457).

ASYNC_MODE instance-attribute

ASYNC_MODE: Literal['sync', 'async']

What mode = "auto" model controllers use (default "sync").

WARN_BLOCKING_MS instance-attribute

WARN_BLOCKING_MS: float | None

Warn when sync code blocks an async operation's event loop longer than this.

ASYNC_FETCH_MODE instance-attribute

ASYNC_FETCH_MODE: Literal['lazy', 'raise']

"raise" makes lazy relation loads fail loudly in async operations (Django 6.1+).

PLUGINS instance-attribute

PLUGINS: Sequence[ControllerPlugin | str]

ControllerPlugin objects (or import paths) applied to every controller.

TENANT_RESOLVER instance-attribute

TENANT_RESOLVER: Callable[[HttpRequest], object] | str

A function of the request returning the tenant (sync or async), or its import path.

TENANT_CONTEXT instance-attribute

TENANT_CONTEXT: object

A RequestContext[User, Tenant] key (or its import path) whose tenant is used.

THROTTLE_RATES instance-attribute

THROTTLE_RATES: Mapping[str, str | None]

Rates for ScopedRateThrottle scopes, e.g. {"uploads": "10/min"}.

THROTTLE_STORAGE instance-attribute

THROTTLE_STORAGE: object

A ThrottleStorage (or its import path) used by throttles without their own storage=; default: cache-based fixed windows.

OBJECT_PERMISSION_BACKEND instance-attribute

OBJECT_PERMISSION_BACKEND: object

An ObjectPermissionBackend (or its import path); default: grants, guardian, Django.

CHECK_APIS instance-attribute

CHECK_APIS: Sequence[str]

NinjaAPI import paths validated by manage.py check.

WEBHOOK_SECRET_KEYS instance-attribute

WEBHOOK_SECRET_KEYS: Sequence[str]

Fernet keys encrypting webhook signing secrets at rest (ninja-devx[crypto]); the first encrypts, all decrypt. Empty: secrets are stored as they are.

class_setting

class_setting(
    cls: type[object], attribute: str, default: T
) -> T

cls.<attribute> when a user class sets it, otherwise the project default.

ninja_devx.serialization.schemas

Schema helpers: typed partial updates and read-only / write-only fields.

Patch[ArticleIn] validates the fields of ArticleIn that were sent (all optional) and gives the handler a PatchData: a dict with .changed and .apply(obj).

ReadOnly / WriteOnly mark fields on one schema; Input[S] hides read-only fields from requests, Output[S] hides write-only fields from responses.

ReadOnly module-attribute

ReadOnly = Annotated[T, READ_ONLY]

A field clients cannot set: hidden and ignored in Input[S].

WriteOnly module-attribute

WriteOnly = Annotated[T, WRITE_ONLY]

A field never returned: hidden and not serialized in Output[S].

PatchData

Bases: dict[str, object], Generic[SchemaT]

The validated fields a client sent for a partial update of SchemaT.

apply

apply(target: T) -> T

Set every sent field on target (no save).

patch_type

patch_type(
    schema: type[BaseModel],
) -> type[PatchData[BaseModel]]

The body type for Patch[schema] (cached; OpenAPI name <Schema>Patch).

ninja_devx.serialization.pydantic

Typed wrapper around pydantic.create_model for schemas built at runtime.

build_schema

build_schema(
    name: str,
    base: type[SchemaT],
    fields: Mapping[str, tuple[object, object]],
) -> type[SchemaT]

A subclass of base named name with fields (name -> (annotation, default)).

ninja_devx.models

Django model discovery for optional core persistence features, and abstract model bases.

SoftDeletable

Bases: Model

deleted_at/deleted_by with the defaults SoftDeleteMixin expects.

A controller using SoftDeleteMixin needs no soft_delete configuration for a model built on this base: rows are marked with the time and the requesting user.

Stamped

Bases: TimeStamped, UserStamped

Timestamps and user stamps together.

TimeStamped

Bases: Model

created_at (indexed) and updated_at, maintained by Django.

UserStamped

Bases: Model

created_by and updated_by, set from the request user by model controllers.

Both are nullable so anonymous writes and deleted users leave None.

ninja_devx.stamps

Abstract model bases for bookkeeping columns: timestamps, user stamps, soft deletion.

Combine them with a model and the CRUD controllers fill the columns::

from ninja_devx.models import SoftDeletable, Stamped

class Post(Stamped, SoftDeletable):
    title = models.CharField(max_length=200)

created_at/updated_at are maintained by Django (auto_now_add/auto_now); created_by/updated_by are set from the request user on create and update, and deleted_at/deleted_by by SoftDeleteMixin. All of them are editable=False, so generated input schemas, scaffolding and drift checks leave them out.

TimeStamped

Bases: Model

created_at (indexed) and updated_at, maintained by Django.

UserStamped

Bases: Model

created_by and updated_by, set from the request user by model controllers.

Both are nullable so anonymous writes and deleted users leave None.

Stamped

Bases: TimeStamped, UserStamped

Timestamps and user stamps together.

SoftDeletable

Bases: Model

deleted_at/deleted_by with the defaults SoftDeleteMixin expects.

A controller using SoftDeleteMixin needs no soft_delete configuration for a model built on this base: rows are marked with the time and the requesting user.

ninja_devx.plugins

Application plugins: bundle cross-cutting setup for a NinjaAPI.

A plugin contributes middleware, error rules and startup work in one object, so an application installs a coherent unit instead of wiring each piece::

class Observability(APIPlugin):
    def middleware(self) -> Sequence[Middleware]:
        return [RequestIDMiddleware(), OpenTelemetryMetricsMiddleware()]

    def errors(self) -> ErrorMap | None:
        return ErrorMap.django_defaults()

    def setup(self, api: NinjaAPI) -> None:
        api.title = "My API"

install(api, [Observability()])

Like use_middleware, install runs before routers are mounted. It is distinct from :class:~ninja_devx.routing.plugins.ControllerPlugin, which extends individual controllers/operations.

APIPlugin

Base class for an application plugin. Override the parts you need.

middleware

middleware() -> Sequence[Middleware]

Middleware applied to every operation of the API.

errors

errors() -> ErrorMap | None

Exception rules installed on the API.

setup

setup(api: NinjaAPI) -> None

Runs after middleware and error rules are installed.

install

install(
    api: NinjaAPI, plugins: Sequence[APIPlugin]
) -> None

Install plugins on api: error rules, middleware, then setup.

Error rules of all plugins are merged into one ErrorMap before installation, so a rule for a subclass wins over a base-class rule regardless of plugin order. Call it before mounting routers.

Parameters:

Name	Type	Description	Default
api	NinjaAPI	The NinjaAPI to configure.	required
plugins	Sequence[APIPlugin]	Plugins applied in order.	required

CRUD

ninja_devx.crud.controllers

Generic model controllers and composable CRUD mixins (sync and async from one source).

The generic arguments are the configuration::

class ArticleController(CRUDController[Article, ArticleOut, ArticleIn]):
    owner_field = "author"
    search_fields = ("title", "body")
    ordering_fields = ("created", "title")

mode picks sync or async operations ("auto" follows NINJA_DEVX["ASYNC_MODE"]). Writes go through a ModelService (service_class) and its repository, so business rules live outside HTTP; the controller only adds what comes from the request (owner, parent) and handles permissions, 404s and responses.

PatchData

Bases: dict[str, object], Generic[SchemaT]

The validated fields a client sent for a partial update of SchemaT.

apply

apply(target: T) -> T

Set every sent field on target (no save).

ModelController

Bases: Controller, Generic[ModelT]

Base for controllers backed by a Django model.

Every hook receives request explicitly, so subclasses stay safe with Scope.SINGLETON. Class attributes left unset fall back to settings.NINJA_DEVX.

mode class-attribute

mode: Literal['sync', 'async', 'auto'] = 'auto'

Which implementation of each operation to register: "auto" follows NINJA_DEVX["ASYNC_MODE"].

lookup_field class-attribute

lookup_field: str = 'pk'

Model field matched by the lookup path segment ("slug", "uuid").

lookup_param class-attribute

lookup_param: str = 'pk'

Name of the lookup path segment: "slug" exposes /{slug}. Declared paths keep using {pk}.

lookup_converter class-attribute

lookup_converter: bool = False

Use Django path converters ({int:pk}): invalid ids get Django's 404 instead of a JSON 422, and static routes of routers mounted later are never shadowed.

owner_field class-attribute

owner_field: str | None = None

Path to the user ("author", "customer__user"): enforced with IsAuthenticated and IsOwner; a direct foreign key is also set on create.

scope_queryset_to_owner class-attribute

scope_queryset_to_owner: bool = False

Also restrict every query (including lists) to the current user's objects.

tenant_field class-attribute

tenant_field: str | None = None

The field pointing at the tenant ("organization", "project__organization"): every query is filtered by the current tenant, which is also set on create.

tenant_resolver class-attribute

tenant_resolver: TenantResolver | None = None

Returns the request's tenant (sync or async); defaults to NINJA_DEVX["TENANT_RESOLVER"]. Assign it with staticmethod(...).

tenant_context class-attribute

tenant_context: object | None = None

A RequestContext[User, Tenant] key resolved from the container; its tenant is used. Defaults to NINJA_DEVX["TENANT_CONTEXT"].

object_permissions class-attribute

object_permissions: ObjectPermissions | None = None

Per-object Django permissions (grants or django-guardian): object checks by HTTP method and, with filter_lists, querysets limited to viewable objects.

sparse_fields class-attribute

sparse_fields: bool = False

Let clients pick response fields with ?fields=id,title (list and retrieve); the output schema needs the FieldVisibility mixin. Expandable relations add ?expand=.

fields_param class-attribute

fields_param: str = 'fields'

Query parameter of sparse_fields.

expand_param class-attribute

expand_param: str = 'expand'

Query parameter listing the Expandable relations to embed.

etag class-attribute

etag: ETag | None = None

Conditional requests: ETag and 304 on reads, If-Match and 412 on writes.

parent class-attribute

parent: Parent | None = None

Nest under a parent from the URL, e.g. Parent(Author, field="author").

service_class class-attribute

service_class: type[object] | None = None

The ModelService subclass handling writes, resolved from the container when there is one. Defaults to a ModelService over a ModelRepository of the model.

validate_model class-attribute

validate_model: bool = True

Run Model.full_clean() before saving (for the default service); errors are 422.

refresh_after_write class-attribute

refresh_after_write: bool = True

Re-fetch through scoped_queryset() after writes so responses see its joins.

optimize_queries class-attribute

optimize_queries: bool | Literal['only'] = True

Join/prefetch what the output schema renders; "only" also restricts columns.

related class-attribute

related: Sequence[str] = ()

Explicit lookups the N+1 planner always loads (("author", "comments__user")), for resolvers or properties it cannot analyse. Also set with @requires_related.

expand_rules class-attribute

expand_rules: Mapping[str, ExpandRule] = MappingProxyType(
    {}
)

How an expanded to-many relation is loaded ({"comments": ExpandRule(limit=5)}); keys must be Expandable fields of the output schema.

openapi_examples class-attribute

openapi_examples: bool = False

Fill an OpenAPI example into the input and output schemas from ninja_devx.testing.sample (see ninja_devx.serialization.examples.with_examples).

authorize_replay

authorize_replay(
    request: HttpRequest,
    operation: OperationInfo,
    arguments: Mapping[str, object],
) -> object

Recheck current ownership, grants, tenancy and deletion before a replay.

check_write_visibility

check_write_visibility(
    request: HttpRequest,
    schema: type[BaseModel] | None,
    sent: object,
) -> None

Reject writes to fields the caller may not set (WriteVisibleTo).

get_queryset

get_queryset(request: HttpRequest) -> QuerySet[ModelT]

Override to scope and optimize queries (tenancy, annotations...).

get_tenant

get_tenant(request: HttpRequest) -> object

The current tenant (resolved once per request); 403 when there is none.

aprepare_request async

aprepare_request(request: HttpRequest) -> None

Load what sync code will need without blocking the event loop: the user and tenant.

scoped_queryset

scoped_queryset(request: HttpRequest) -> QuerySet[ModelT]

get_queryset() restricted to the tenant, parent and owner, with N+1 optimizations.

get_object

get_object(
    request: HttpRequest,
    lookup: object,
    *,
    lock: bool = False,
) -> ModelT

Fetch one object (404 when missing) and enforce object permissions.

object_etag

object_etag(request: HttpRequest, instance: ModelT) -> str

The entity tag of instance (etag must be set).

conditional_object

conditional_object(
    request: HttpRequest, instance: ModelT
) -> ModelT | HttpResponseBase

instance, or a 304 when the client's If-None-Match still matches it.

check_preconditions

check_preconditions(
    request: HttpRequest, instance: ModelT
) -> None

Before a write: 412 when If-Match is stale, 428 when it is required and missing.

written

written(request: HttpRequest, instance: ModelT) -> ModelT

After a write: send the new ETag.

refresh

refresh(request: HttpRequest, instance: ModelT) -> ModelT

Verify persisted scope and optionally reload the result through its write database.

write_database

write_database(request: HttpRequest) -> str

Operation alias, explicit queryset alias, or the model's write router.

get_service

get_service(request: HttpRequest) -> ModelService[ModelT]

The write service: service_class from this call's container, or the default.

context_data

context_data(request: HttpRequest) -> dict[str, object]

Fields set from the request on create: tenant, owner, parent and user stamps.

update_context_data

update_context_data(
    request: HttpRequest,
) -> dict[str, object]

Fields set from the request on every update: updated_by for stamped models.

perform_update

perform_update(
    request: HttpRequest,
    instance: ModelT,
    data: Mapping[str, object],
) -> ModelT

Update through the service with the data plus update_context_data.

on_change

on_change(
    request: HttpRequest,
    instance: ModelT,
    changes: Mapping[str, tuple[object, object]],
) -> None

After a successful update, inside the transaction: changes maps field name to (old, new) (see changed_fields). No-op by default.

check_input_schema classmethod

check_input_schema(hook: str) -> None

Fail at startup when the input schema does not match the model.

Skipped when hook or service_class is customized, since that code may map the fields itself.

ListConfig

Bases: ModelController[ModelT], Generic[ModelT, OutT]

Filtering, search, ordering, pagination and selectors for list endpoints.

filter_schema class-attribute

filter_schema: type[FilterSchema] | None = None

An explicit Ninja FilterSchema for GET / (instead of generated filters).

search_fields class-attribute

search_fields: Sequence[str] = ()

Fields searched with icontains by the search_param query parameter.

search_backend class-attribute

search_backend: SearchBackend[Model] | None = None

Replace the default icontains search (for example PostgresSearch()).

search_param class-attribute

search_param: str = 'search'

Name of the search query parameter.

filter_fields class-attribute

filter_fields: FilterFields = MappingProxyType({})

Generated typed filters: {"status": ("exact",), "created": ("gte", "lte")}.

filterset_class class-attribute

filterset_class: type[FilterSetLike] | None = None

A django-filter FilterSet whose filters become query parameters of GET / (pip install ninja-devx[filters]); it runs with the request.

ordering_fields class-attribute

ordering_fields: Sequence[str] = ()

Fields the client may order by (?ordering=-created), validated as an enum.

ordering_param class-attribute

ordering_param: str = 'ordering'

Name of the ordering query parameter.

default_ordering class-attribute

default_ordering: Sequence[str] = ()

Ordering when the client sends none; must be among ordering_fields.

pagination_class class-attribute

pagination_class: type[PaginationBase] | None = None

A Ninja pagination class (or CursorPagination); defaults to NINJA_DEVX["PAGINATION_CLASS"].

pagination_options class-attribute

pagination_options: Mapping[str, object] = MappingProxyType(
    {}
)

Keyword arguments for the pagination class ({"page_size": 50}).

selector_class class-attribute

selector_class: type[Selector[Never, Model]] | None = None

A selector (resolved from the container when there is one) replacing get_queryset for lists. Querysets it returns are still ordered, paginated and optimized.

ListMixin

Bases: ListConfig[ModelT, OutT], Generic[ModelT, OutT]

GET / with filtering, search, ordering and pagination.

RetrieveMixin

Bases: ModelController[ModelT], Generic[ModelT, OutT]

GET /{pk}.

CreateHooks

Bases: ModelController[ModelT], Generic[ModelT, InT]

perform_create typed with the input schema, shared by every create operation.

perform_create

perform_create(
    request: HttpRequest, payload: InT
) -> ModelT

Create through the service with the payload plus the owner and parent.

CreateMixin

Bases: CreateHooks[ModelT, InT], Generic[ModelT, OutT, InT]

POST / returning 201.

UpdateMixin

Bases: ModelController[ModelT], Generic[ModelT, OutT, InT]

PUT /{pk} (full) and PATCH /{pk} (only the fields sent).

DestroyMixin

Bases: ModelController[ModelT], Generic[ModelT]

DELETE /{pk} returning 204.

ReadOnlyModelController

Bases: ListMixin[ModelT, OutT], RetrieveMixin[ModelT, OutT], Generic[ModelT, OutT]

GET / and GET /{pk}.

CRUDController

Bases: ListMixin[ModelT, OutT], RetrieveMixin[ModelT, OutT], CreateMixin[ModelT, OutT, InT], UpdateMixin[ModelT, OutT, InT], DestroyMixin[ModelT], Generic[ModelT, OutT, InT]

List, retrieve, create, update, partial update and destroy.

ninja_devx.crud.async_controllers

Async CRUD controllers: the same classes as the sync ones, with mode = "async".

Every CRUD mixin carries both implementations (see async_variant); these aliases just pick the async ones, so routes, operation ids and hooks are identical.

ninja_devx.crud.annotations

Annotations resolved per controller when its router is built.

Lookup module-attribute

Lookup: TypeAlias = Annotated[
    int | str | UUID, _LookupMarker()
]

Path parameter typed like the controller's lookup_field (the primary key by default).

Instance module-attribute

Instance: TypeAlias = Annotated[ModelT, _InstanceMarker()]

A model instance loaded from the {pk} path segment, with 404 and object permissions.

Locked module-attribute

Locked: TypeAlias = Annotated[
    ModelT, _InstanceMarker(lock=True)
]

Like Instance but loaded with select_for_update() (use on atomic=True operations).

Filters module-attribute

Filters: TypeAlias = Annotated[
    FilterSchema, _FiltersMarker()
]

Query parameters from filter_schema or search_fields/filter_fields.

Ordering module-attribute

Ordering: TypeAlias = Annotated[
    OrderingSchema, _OrderingMarker()
]

Query parameter ordering restricted to the controller's ordering_fields.

OrderingSchema

Bases: Schema

Ordering chosen by the client; the base class exposes no parameters.

controller_lookup_type

controller_lookup_type(
    controller: type[object],
) -> type[object]

The Python type of the controller's lookup_field.

lookup_param

lookup_param(controller: type[object]) -> str

The URL name of the lookup segment (lookup_param, "pk" by default).

with_path_converter

with_path_converter(
    path: str, controller: type[object]
) -> str

/{pk} -> /{<lookup_param>}, or /{int:<lookup_param>} with lookup_converter.

Without a converter an invalid id reaches Ninja and gets a JSON 422; with one, Django answers 404 itself, which keeps routes of routers mounted later (/posts/me) reachable.

query_schema

query_schema(schema: type[Schema]) -> object

A query-parameter schema; schemas without required fields may be omitted entirely.

ordering_schema

ordering_schema(
    controller: type[object],
) -> type[OrderingSchema]

The ordering schema of a controller, cached so OpenAPI component names stay stable.

ninja_devx.crud.bulk

Bulk create, update and delete with a size limit, in one transaction each.

BulkResultOut module-attribute

BulkResultOut = Annotated[BaseModel, _BulkResultMarker()]

{"results": [{"index", "status", "data"} | {"index", "status", "errors"}]}: the body of a partial-success (207) bulk response, its data typed like the controller's output.

BulkPatch

Bases: Schema, Generic[SchemaT]

{"pks": [...], "data": {...}}: apply the same partial update to many objects.

BulkDelete

Bases: Schema

{"pks": [...]}.

BulkErrorDetail

Bases: Schema

One item of a bulk 207 entry's errors, in Ninja's validation error shape.

type instance-attribute

type: str

Ninja's validation error type ("missing", "value_error", ...).

loc instance-attribute

loc: list[int | str]

Field path within the item, as Ninja reports it (without the item's own index).

msg instance-attribute

msg: str

Human-readable message.

BulkCreateMixin

Bases: _BulkBase[ModelT], CreateHooks[ModelT, InT], Generic[ModelT, OutT, InT]

POST /bulk: create many objects (validation and signals run per object).

bulk_partial class-attribute

bulk_partial: bool = False

Create each item in its own savepoint: failures become 207 entries (results) instead of rolling back the whole request.

BulkUpdateMixin

Bases: _BulkBase[ModelT], Generic[ModelT, OutT, InT]

POST /bulk-update: apply one partial update to many objects.

bulk_partial class-attribute

bulk_partial: bool = False

Update each item in its own savepoint: an unknown pk or a per-item failure becomes a 207 entry (results) instead of rolling back the whole request.

BulkDestroyMixin

Bases: _BulkBase[ModelT], Generic[ModelT]

POST /bulk-delete: delete many objects (perform_destroy per object).

ninja_devx.crud.soft_delete

Soft deletion: hide instead of delete, with a restore endpoint.

Nothing is hard-coded: the field, the values that mean "deleted" and "active", who deleted it, and when, are configured per controller::

class PostController(SoftDeleteMixin[Post, PostOut], CRUDController[Post, PostOut, PostIn]):
    soft_delete = SoftDelete("deleted_at")                        # nullable DateTimeField
    soft_delete = SoftDelete("is_deleted")                        # BooleanField
    soft_delete = SoftDelete("is_active", deleted=False, active=True)
    soft_delete = SoftDelete("status", deleted="archived", active="published")
    soft_delete = SoftDelete("is_deleted", deleted_at="removed_on", deleted_by="removed_by")

The restore route is POST /{pk}/restore; rename or remove it with routes.

SoftDelete dataclass

How a model marks deleted rows.

deleted is the value (or a zero-argument callable, like timezone.now) written on delete; active is the value of rows that are not deleted, written on restore. Both are inferred for a BooleanField (True/False) and a nullable DateTimeField (now/None); other fields need them explicitly.

field class-attribute instance-attribute

field: str = 'deleted_at'

The field marking deleted rows.

deleted class-attribute instance-attribute

deleted: object = INFER

Value written on delete (or a zero-argument callable); inferred for boolean and nullable datetime fields.

active class-attribute instance-attribute

active: object = INFER

Value of rows that are not deleted, written on restore; inferred like deleted.

deleted_at class-attribute instance-attribute

deleted_at: str | None = None

Also set this DateTimeField to now on delete (and clear it on restore).

deleted_by class-attribute instance-attribute

deleted_by: str | None = None

Also set this foreign key to the current user on delete (and clear it on restore). Defaults to "deleted_by" for models built on SoftDeletable.

SoftDeleteMixin

Bases: ModelController[ModelT], Generic[ModelT, OutT]

Marks objects deleted instead of deleting them and adds POST /{pk}/restore.

List it before the CRUD base so its perform_destroy wins: class X(SoftDeleteMixin[...], CRUDController[...]).

soft_delete class-attribute

soft_delete: SoftDelete | str = SoftDelete()

A SoftDelete or just the field name.

soft_delete_cascade class-attribute

soft_delete_cascade: Sequence[str] = ()

Related accessor names marked deleted/restored with this object (("comments", "attachments")). Only the configured marker field is cascaded.

queryset_with_deleted

queryset_with_deleted(
    request: HttpRequest,
) -> QuerySet[ModelT]

scoped_queryset including deleted objects (parent, owner and tenant still apply).

soft_delete_unique

soft_delete_unique(
    model: type[Model],
    *fields: str,
    config: SoftDelete | None = None,
    name: str | None = None,
) -> UniqueConstraint

A partial UniqueConstraint that only applies to active (not deleted) rows.

Add it to the model so a deleted row's value can be reused::

class Post(models.Model):
    slug = models.SlugField()
    deleted_at = models.DateTimeField(null=True)

    class Meta:
        constraints = [soft_delete_unique(Post, "slug")]

Parameters:

Name	Type	Description	Default
model	type[Model]	The model (used to resolve the marker field and its active value).	required
fields	str	The constrained fields.	()
config	SoftDelete | None	The SoftDelete configuration (default: deleted_at).	None
name	str | None	Constraint name (default derived from the table and fields).	None

ninja_devx.crud.nested

Nested resources: /authors/{author_pk}/articles.

Parent dataclass

Scope a model controller to a parent object taken from the URL.

Parent(Author, field="author") expects the router to be mounted under a prefix containing {author_pk}; every operation then 404s for unknown authors, lists only that author's objects and assigns the author on create.

model instance-attribute

model: type[Model]

The parent model.

field instance-attribute

field: str

The child's foreign key to the parent.

lookup_field class-attribute instance-attribute

lookup_field: str = 'pk'

Parent field matched by the URL segment.

param class-attribute instance-attribute

param: str | None = None

Name of the URL segment (default <field>_pk).

tenant_field class-attribute instance-attribute

tenant_field: str | None = None

Only find parents of the current tenant ("organization"): other tenants' parents 404.

get_parent

get_parent(request: HttpRequest) -> Model

The parent object of a nested operation (loaded before the controller runs).

ninja_devx.crud.nested_writes

Write a parent and its child collections together, in one request and transaction.

::

class OrderController(
    NestedWritesMixin[Order, OrderIn], CRUDController[Order, OrderOut, OrderIn]
):
    nested = {"items": Nested(OrderItem, "order", OrderItemIn)}

OrderIn must declare items: list[OrderItemIn]; OrderItem.order is the foreign key back to Order. On create, the parent is saved and then every item, all inside the operation's write_scope. On update (PUT/PATCH), an item sent with its key field (default "id") is matched against the parent's existing children and updated in place; an item sent without it is created; existing children whose key does not appear in the payload are deleted unless Nested(..., remove_missing=False). A PATCH that does not send the field at all leaves the children untouched. Validation errors on an item are reported at <field>.<index>.<...>, e.g. items.2.quantity.

Nested dataclass

One child collection written alongside its parent.

Nested(OrderItem, "order", OrderItemIn): OrderItem.order is the foreign key to the parent, OrderItemIn validates each item.

model instance-attribute

model: type[Model]

The child model.

field instance-attribute

field: str

The child's foreign key to the parent.

schema instance-attribute

schema: type[BaseModel]

Input schema validating each item.

key class-attribute instance-attribute

key: str = 'id'

Field matching a payload item against an existing child on update.

remove_missing class-attribute instance-attribute

remove_missing: bool = True

On update, delete existing children whose key is absent from the payload.

NestedWritesMixin

Bases: ModelController[ModelT], Generic[ModelT, InT]

Adds child collections to create and update, declared with nested.

List it before the CRUD base so its perform_create/perform_update win, like SoftDeleteMixin.

nested class-attribute

nested: Mapping[str, Nested] = MappingProxyType({})

Child collections keyed by their field on the input schema.

nested_config classmethod

nested_config() -> Mapping[str, Nested]

nested, validated once against the model and input schema.

perform_create

perform_create(
    request: HttpRequest, payload: InT
) -> ModelT

Create the parent, then every declared child collection, one transaction.

perform_update

perform_update(
    request: HttpRequest,
    instance: ModelT,
    data: Mapping[str, object],
) -> ModelT

Update the parent's own fields, then sync any child collections present in data.

ninja_devx.crud.transitions

State transitions exposed as API operations, with permissions, guards and a listing route.

Declare the allowed moves; each becomes POST /{pk}/<name>::

class ArticleController(TransitionsMixin[Article, ArticleOut], CRUDController[...]):
    state_field = "status"
    transitions = {
        "publish": Transition(source=("draft",), target="published", permissions=[IsStaff()]),
        "archive": Transition(source=("published",), target="archived"),
    }

GET /{pk}/transitions lists the names allowed from the object's current state for the caller (permissions and guards evaluated, without performing anything). Rename or disable routes with routes (operation names are transition_<name> and transitions_list).

This stays API-level: it does not integrate with model-level FSMs (django-fsm-2 and similar). Call a @transition-decorated model method from Transition.on_transition or on_transition() if a project already has one.

InvalidTransition

Bases: Conflict

The object's current state is not a source state for this transition.

Transition dataclass

One named move from a set of source states to a target state.

source instance-attribute

source: Sequence[str]

States the object must be in for this transition to apply.

target instance-attribute

target: str

State written when the transition succeeds.

permissions class-attribute instance-attribute

permissions: Sequence[AnyPermission] = ()

Added to the controller's permissions for this transition's route only.

guard class-attribute instance-attribute

guard: Callable[[HttpRequest, Model], bool] | None = None

Extra precondition beyond the source state; False also answers 409.

on_transition class-attribute instance-attribute

on_transition: (
    Callable[[HttpRequest, Model], None] | None
) = None

Called after the new state is saved, inside the write transaction.

TransitionsMixin

Bases: ModelController[ModelT], Generic[ModelT, OutT]

Adds POST /{pk}/<name> for each declared transition and GET /{pk}/transitions.

state_field class-attribute

state_field: str = 'status'

The CharField holding the object's state.

transitions class-attribute

transitions: Mapping[str, Transition] = MappingProxyType({})

Transition name to :class:Transition.

transition_allowed

transition_allowed(
    request: HttpRequest,
    instance: ModelT,
    config: Transition,
) -> bool

Whether instance's current state and config.guard allow the transition.

Does not check permissions; see allowed_transitions for that.

perform_transition

perform_transition(
    request: HttpRequest, instance: ModelT, name: str
) -> ModelT

Write config.target to state_field and run the transition hooks.

Raises InvalidTransition (409) when the current state or the guard refuses.

on_transition

on_transition(
    request: HttpRequest, instance: ModelT, name: str
) -> None

Called after any transition, after Transition.on_transition. No-op by default.

allowed_transitions

allowed_transitions(
    request: HttpRequest, instance: ModelT
) -> list[str]

Names of the transitions instance's state, guards and permissions allow.

ninja_devx.crud.meta

GET /meta describing a model controller's fields, for building forms and admin UIs.

::

class NoteController(MetaMixin[Note], CRUDController[Note, NoteOut, NoteIn]):
    filter_fields = {"status": ("exact",)}
    ordering_fields = ("priority",)
    search_fields = ("text",)

The response lists every input and output schema field with its type name, whether it is required or read-only, its max length and its choices (from the matching model field's choices, or from an enum.Enum schema type), plus the controller's filter_fields, ordering_fields and search_fields names. A field is read-only when the output schema carries it but the input schema does not. The structure needs no database access, so it is computed once and cached per controller class.

ChoiceOut

Bases: Schema

One value/label choice.

FieldMeta

Bases: Schema

What a form or admin UI needs to know about one schema field.

ControllerMeta

Bases: Schema

The body of GET /meta.

MetaMixin

Bases: ModelController[ModelT], Generic[ModelT]

Adds GET /meta, describing the input and output schemas.

ninja_devx.crud.aggregates

GET /stats grouping and aggregating a model controller's scoped list queryset.

::

class OrderController(
    AggregateMixin[Order, OrderOut], CRUDController[Order, OrderOut, OrderIn]
):
    aggregate_fields = ("status", "region")
    aggregate_metrics = {"total": Sum("amount"), "count": Count("id")}

GET /stats?group_by=status&metrics=total&metrics=count groups the same queryset the list operation would serve (tenant, owner, soft deletion and search/filter_fields all apply) and answers [{"status": "done", "total": "120.00", "count": 3}, ...]. Omitting group_by aggregates the whole queryset into one row. group_by is restricted to aggregate_fields and metrics to the keys of aggregate_metrics (default ["count"]); anything else is a 422.

AggregateParams module-attribute

AggregateParams: TypeAlias = Annotated[
    Schema, _AggregateQueryMarker()
]

Query parameters group_by (from aggregate_fields) and metrics (from aggregate_metrics, default ["count"]).

AggregateMixin

Bases: ListConfig[ModelT, OutT], Generic[ModelT, OutT]

GET /stats: grouped counts and sums over the controller's scoped queryset.

aggregate_fields class-attribute

aggregate_fields: Sequence[str] = ()

Allow-list of model fields group_by may name.

aggregate_metrics class-attribute

aggregate_metrics: Mapping[str, Aggregate] = (
    MappingProxyType({"count": Count("pk")})
)

Allow-list of aggregate expressions metrics may name, by the name clients use.

aggregate_schema_for

aggregate_schema_for(
    controller: type[object],
) -> type[Schema]

The group_by/metrics query schema of a controller (cached).

ninja_devx.crud.optimization

Derive select_related/prefetch_related from output schemas to avoid N+1 queries.

Forward foreign keys rendered as nested schemas are joined. Many-to-many and reverse relations are prefetched with a Prefetch whose queryset joins their nested foreign keys, so comments: list[CommentOut] with CommentOut.author: UserOut costs one query per relation, not one per level.

optimize_queryset(queryset, schema, only=True) also restricts the selected columns to the fields the schema reads; it is skipped automatically when the schema reads anything that is not a model field (resolvers, properties), since those could load deferred columns one row at a time.

ExpandRule shapes an expanded to-many relation (filter, order, a per-parent cap) through a Prefetch queryset.

ExpandRule dataclass

How ?expand= loads a to-many relation: filtered, ordered and capped per parent.

::

class ArticleController(ReadOnlyModelController[Article, ArticleOut]):
    expand_rules = {
        "comments": ExpandRule(filter=Q(published=True), order_by=("-created",), limit=5),
    }

Rule keys must name an Expandable field of the output schema (ControllerConfigError otherwise). limit ranks rows per parent with a window function (a QUALIFY-style filter on Django 5.0+) or, on Django 4.2, an equivalent correlated subquery; it is skipped, with ninja_devx.W007, when the relation is not a plain reverse foreign key (a many-to-many, for example).

filter class-attribute instance-attribute

filter: Q | None = None

Restricts the related rows, applied before order_by and limit.

order_by class-attribute instance-attribute

order_by: tuple[str, ...] = ()

Ordering applied before limit; defaults to the related model's Meta.ordering.

limit class-attribute instance-attribute

limit: int | None = None

Rows kept per parent object.

QueryPlan dataclass

How to load model for schema: joins, prefetches (each with its own plan).

lookups

lookups() -> tuple[tuple[str, ...], tuple[str, ...]]

Flat (select_related, prefetch_related) lookups, for inspection.

requires_related

requires_related(*lookups: str) -> Callable[[T], T]

Declare relations a schema resolver reads, so the planner loads them explicitly.

@requires_related("author", "comments__user") on a resolve_<field> method adds the lookups to the query plan even though a resolver body cannot be analysed.

query_plan

query_plan(
    model: type[Model],
    schema: type[BaseModel],
    expand: frozenset[str] = frozenset(),
    hints: Sequence[str] = (),
) -> QueryPlan

The (cached) loading plan for serializing model instances with schema.

Expandable relations are loaded only when named in expand. hints are extra select_related/prefetch_related lookups applied on top of the derived plan.

related_lookups

related_lookups(
    model: type[Model], schema: type[BaseModel]
) -> tuple[tuple[str, ...], tuple[str, ...]]

(select_related, prefetch_related) lookups to serialize model with schema.

resolve_expand_attribute

resolve_expand_attribute(
    schema: type[BaseModel], name: str
) -> str

The model attribute an Expandable schema field name reads (its alias, if any).

expand_limit_target

expand_limit_target(
    model: type[Model], lookup: str
) -> tuple[str, type[Model]] | None

(partition column, related model) for capping lookup per parent, or None when lookup is not a plain reverse foreign key of model.

ninja_devx.crud.pagination

Pagination built on Ninja's paginators: cursor and limit/offset.

CursorPagination is the cursor paginator that follows the list's ordering.

Ninja's CursorPagination orders by a fixed tuple chosen when the view is decorated. This subclass takes the ordering of each request instead (?ordering= restricted to ordering_fields, else default_ordering, else the model's Meta.ordering, else -pk), adds the primary key as a tiebreaker, and rejects a cursor created for a different ordering::

class EventController(ReadOnlyModelController[Event, EventOut]):
    pagination_class = CursorPagination
    pagination_options = {"page_size": 50}
    ordering_fields = ("created", "priority")
    default_ordering = ("-created",)

Cursors compare the first ordering field, so ordering fields must not be nullable; that is checked at startup.

CursorPagination

Bases: CursorPagination

Cursor

Bases: Cursor

k class-attribute instance-attribute

k: str | None = None

Fingerprint of the ordering the cursor was created for.

LimitOffsetPagination

Bases: LimitOffsetPagination

?limit=&offset= pages with stable ordering, links and a configurable count.

::

class EventController(ReadOnlyModelController[Event, EventOut]):
    pagination_class = LimitOffsetPagination
    pagination_options = {"limit": 50, "max_limit": 200, "count": False}

Compared with Ninja's LimitOffsetPagination:

• pages never overlap or skip rows: the primary key is added to the ordering;
• next/previous links keep the other query parameters;
• limit above max_limit is clamped instead of rejected, and max_offset turns deep offsets (slow on large tables) into a 422 suggesting cursor pagination.

count picks how the total is produced (the response body's count is always a plain int or null):

• True (default): an exact COUNT(*).
• False: no count query (count is null); one extra row is fetched instead, to know whether there is a next page.
• "estimate": on PostgreSQL, pg_class.reltuples (instant, approximate) for an unfiltered queryset; an exact count otherwise, and on every other database.
• an integer N: exact while there are at most N rows; beyond that, count is N and PaginationHeadersMiddleware sends X-Total-Count: N+ instead of N.

ninja_devx.crud.auto

CRUD controllers whose schemas are generated from the model by Ninja's create_schema.

::

class PostController(AutoCRUDController[Post]):
    read_only_fields = ("author", "created")
    write_only_fields = ("secret",)

• The output schema (PostOut) has schema_fields minus schema_exclude and write_only_fields.
• The input schema (PostIn) also leaves out the primary key, non-editable fields (auto_now, GeneratedField) and read_only_fields. Fields with a db_default are optional; left out, the database supplies the value.
• Everything else (filters, permissions, tenancy, pagination...) is configured as on CRUDController. Switch to explicit schemas once they need custom fields.

AutoSchemas

Which model fields the generated schemas contain.

schema_fields class-attribute

schema_fields: Sequence[str] | Literal["__all__"] = (
    "__all__"
)

Model fields in the schemas (field names; "__all__": every concrete and m2m field).

schema_exclude class-attribute

schema_exclude: Sequence[str] = ()

Fields left out of both schemas.

read_only_fields class-attribute

read_only_fields: Sequence[str] = ()

Fields in responses only. The primary key and non-editable fields always are.

write_only_fields class-attribute

write_only_fields: Sequence[str] = ()

Fields in requests only (password).

AutoReadOnlyController

Bases: ReadOnlyModelController[ModelT, BaseModel], AutoSchemas, Generic[ModelT]

GET / and GET /{pk} with a generated output schema.

AutoCRUDController

Bases: CRUDController[ModelT, BaseModel, BaseModel], AutoSchemas, Generic[ModelT]

Full CRUD with generated <Model>Out / <Model>In schemas.

model_schemas

model_schemas(
    model: type[Model],
    *,
    fields: Sequence[str] | Literal["__all__"] = "__all__",
    exclude: Sequence[str] = (),
    read_only: Sequence[str] = (),
    write_only: Sequence[str] = (),
) -> tuple[type[Schema], type[Schema]]

(<Model>Out, <Model>In) generated with Ninja's create_schema (cached).

Parameters:

Name	Type	Description	Default
model	type[Model]	The Django model.	required
fields	Sequence[str] | Literal['__all__']	Field names to include, or "__all__".	'__all__'
exclude	Sequence[str]	Field names left out of both schemas.	()
read_only	Sequence[str]	Field names only in the output schema.	()
write_only	Sequence[str]	Field names only in the input schema.	()

ninja_devx.crud.transfer

CSV and JSON Lines export and import for model controllers.

::

class ContactController(ExportMixin[Contact, ContactOut], ImportMixin[Contact, ContactIn],
                        CRUDController[Contact, ContactOut, ContactIn]):
    export_formats = ("csv", "jsonl")

• GET /export?format=csv streams every object the list operation would return (same filters, ordering, scoping and field visibility), without pagination.
• POST /import takes a CSV or JSONL upload, validates every row with the input schema and creates the objects through perform_create in one transaction: either all rows are imported or none (?dry_run=true only validates). Errors are 422 with loc: ["file", <row>, <field>].

ExportMixin

Bases: ListConfig[ModelT, OutT], Generic[ModelT, OutT]

GET /export: the filtered list as CSV or JSON Lines, streamed.

export_formats class-attribute

export_formats: Sequence[ExportFormat] = ('csv', 'jsonl')

Formats clients may ask for; the first is the default.

export_filename class-attribute

export_filename: str | None = None

Download name without extension (default: the model's plural name).

csv_escape_formulas class-attribute

csv_escape_formulas: bool = True

Prefix text cells starting with = + - @ with ' so spreadsheets don't run them.

export_sensitive class-attribute

export_sensitive: bool = False

Include Sensitive output fields unmasked (default: exported as "***").

ImportMixin

Bases: CreateHooks[ModelT, InT], Generic[ModelT, InT]

POST /import: create objects from a CSV or JSON Lines file, all or nothing.

max_import_bytes class-attribute

max_import_bytes: int = 10000000

Reject files larger than this byte limit before parsing or starting writes.

max_import_rows class-attribute

max_import_rows: int = 10000

Larger files are rejected with 413.

max_import_errors class-attribute

max_import_errors: int = 50

Stop validating after this many errors.

ninja_devx.crud.sharing

Share objects over the API: list, grant and revoke object permissions per user or group.

::

class DocumentController(ObjectSharingMixin[Document], CRUDController[Document, Out, In]):
    object_permissions = ObjectPermissions()
    shareable_permissions = ("view", "change")     # codename prefixes clients may grant

Routes (rename or disable them with routes):

• GET /{pk}/permissions — who holds what on the object;
• PUT /{pk}/permissions — set a user's or group's permissions (replaces them);
• POST /{pk}/permissions/revoke — remove a user's or group's permissions.

Managing sharing requires sharing_permission on the object (change by default).

ObjectSharingMixin

Bases: ModelController[ModelT], Generic[ModelT]

Endpoints managing object permissions (grants backend or django-guardian).

shareable_permissions class-attribute

shareable_permissions: tuple[str, ...] = (
    "view",
    "change",
    "delete",
)

Permission actions clients may grant (view means <app>.view_<model>).

sharing_permission class-attribute

sharing_permission: str = 'change'

Action the caller needs on the object to see or change its sharing.

validate_holder

validate_holder(
    request: HttpRequest, obj: ModelT, holder: object
) -> None

Override to restrict who objects may be shared with (raise HttpError(422)), e.g. members of the same workspace. Revoking is always allowed.

ninja_devx.crud.search

Pluggable full-text search for list endpoints.

Without a backend, search_fields become icontains lookups in the generated filter schema. Set search_backend on a controller to replace that, for example with PostgreSQL's SearchVector; the search parameter stays documented either way::

from ninja_devx.crud import PostgresSearch

class ArticleController(CRUDController[Article, ArticleOut, ArticleIn]):
    search_fields = ("title", "body")
    search_backend = PostgresSearch(config="english")

SearchBackend

Bases: Protocol[ModelT]

Turns a search term and field names into a filtered queryset (lazy).

IContainsSearch

Case-insensitive substring match across the fields, as a backend.

Equivalent to the built-in behaviour; a starting point for custom backends.

PostgresSearch

PostgreSQL full-text search across the fields with a shared language config.

The vector is computed per query. Large tables need a stored SearchVectorField with a GIN index and a backend that filters on it.

Parameters:

Name	Type	Description	Default
config	str	to_tsvector/to_tsquery language configuration.	'english'

ninja_devx.crud.fields

Django model field introspection shared by lookups, filters, nesting and scaffolding.

resolve_field

resolve_field(model: type[Model], path: str) -> ModelField

The field at path ("pk", "title", "author__username").

field_type

field_type(
    field: ModelField, *, choices: bool = False
) -> object

The Python type values of field have; relations use the target's key type.

With choices=True a field with choices becomes Literal[...] of its values.

lookup_type

lookup_type(
    model: type[Model], lookup_field: str
) -> type[object]

The Python type of model.<lookup_field> (used for {pk} path parameters).

ninja_devx.crud.filters

Filter schemas generated from search_fields, filter_fields and filterset_class.

FilterFields module-attribute

FilterFields = Mapping[str, Sequence[str]]

{"published": ("exact",), "created": ("gte", "lte"), "tags": ("in",)}.

filter_schema_for

filter_schema_for(
    controller: type[object],
) -> type[FilterSchema]

filter_schema, or one generated from search_fields/filter_fields (cached).

ninja_devx.crud.shaping

?fields= and ?expand= query parameters for model controllers.

partial_schema

partial_schema(schema: type[BaseModel]) -> type[BaseModel]

schema with every field optional in OpenAPI (<Name>Partial), for responses shaped by ?fields=. Validation and serialization are unchanged.

ninja_devx.crud.persistence

Model persistence: re-exports from ninja_devx.layers.persistence, plus changed_fields.

model_field

model_field(
    model: type[Model], name: str
) -> models.Field[object, object] | None

The concrete or many-to-many field called name (or whose attname is name).

save_instance

save_instance(
    instance: ModelT,
    data: Mapping[str, object],
    *,
    validate: bool = True,
    using: str | None = None,
) -> ModelT

Assign data to instance, validate, save and set many-to-many relations.

Foreign keys accept an instance or a primary key. On a new instance, None for a non-null field with db_default leaves the value to the database. Model.full_clean errors are raised as ValidationFailed. Runs in one transaction (a savepoint when nested).

validation_failed

validation_failed(exc: ValidationError) -> ValidationFailed

A Django ValidationError as a domain ValidationFailed (HTTP 422 when mapped).

changed_fields

changed_fields(
    instance: Model, data: Mapping[str, object]
) -> dict[str, tuple[object, object]]

{field: (old, new)} for the keys in data that differ from instance.

Call before writing data to instance: old is read from instance as it stands, new is the incoming value. Foreign keys are compared by primary key; many-to-many fields are not compared (call this before .set() has any effect anyway).

Parameters:

Name	Type	Description	Default
instance	Model	The object about to be updated.	required
data	Mapping[str, object]	The fields being written.	required

Layers

ninja_devx.layers.errors

Domain errors that carry their HTTP meaning without importing any web code.

Raise them from services, repositories and policies; ninja_devx.http.errors.ErrorMap turns anything with http_status and code into a response. Your own exceptions can do the same by defining those two class attributes (see HttpMappable).

HttpMappable

Bases: Protocol

An exception that knows its HTTP status and machine-readable code.

http_status class-attribute

http_status: int

HTTP status of the response.

code class-attribute

code: str

Machine-readable error code.

DomainError

Bases: Exception

Base class for business errors: 400 unless a subclass says otherwise.

http_status class-attribute

http_status: int = 400

HTTP status when mapped (400 unless overridden).

code class-attribute

code: str = 'domain_error'

Machine-readable error code.

default_message class-attribute

default_message: str | None = None

detail when none is given, translated with gettext (default: the docstring).

__init__

__init__(message: str = '', **details: JSON) -> None

Parameters:

Name	Type	Description	Default
message	str	The detail (default: default_message, else the docstring).	''
details	JSON	Extra JSON fields merged into the body.	{}

NotFound

Bases: DomainError

Not found.

Conflict

Bases: DomainError

The request conflicts with the current state.

PermissionDenied

Bases: DomainError

You do not have permission to perform this action.

ValidationFailed

Bases: DomainError

Validation failed.

ninja_devx.layers.repository

Repositories: persistence behind a small, HTTP-free interface.

Repository

Bases: Protocol[ModelT]

Loads and stores one aggregate type. Implement it however your project stores data.

transaction

transaction() -> AbstractContextManager[object]

The unit of work services wrap writes in (transaction.atomic for Django).

ModelRepository

Bases: Generic[ModelT]

The Django ORM repository, sync and async.

Subclass with the model as argument (class PostRepository(ModelRepository[Post])) or pass model=. Async methods use the async ORM for reads and one thread hop (with the transaction) for writes.

queryset

queryset() -> QuerySet[ModelT]

Override to add default filters or joins for every read.

ninja_devx.layers.services

Services: business operations over repositories, callable from sync and async code.

ModelService

Bases: Generic[ModelT]

Create, update and delete through a repository; override to add business rules.

CRUDController sends its writes here when service_class is set::

class PostService(ModelService[Post]):
    @dual
    def create(self, data: Mapping[str, object]) -> Post:
        return super().create({**data, "slug": slugify(data["title"])})

Dependencies come from the container (PostService(repository: PostRepository)); without a registered repository a ModelRepository for the model is used.

ninja_devx.layers.dual

Write a method once, call it from sync and async code.

::

class OrderService:
    @dual
    def place(self, command: PlaceOrder) -> Order: ...

service.place(command)            # sync
await service.place.a(command)    # async: one thread hop, or the native implementation

@place.native
async def _place_async(self, command: PlaceOrder) -> Order: ...

Django's ORM is synchronous underneath, so the default .a runs the sync body in a single sync_to_async(thread_sensitive=True) hop. Provide native only when the async version really avoids blocking (HTTP calls, asyncio libraries...).

dual

Bases: Generic[S, P, R]

A method with a sync implementation and an optional native async one.

native

native(
    function: Callable[Concatenate[S, P], Awaitable[R]],
) -> Callable[Concatenate[S, P], Awaitable[R]]

Register the native async implementation used by .a.

BoundDual

Bases: Generic[P, R]

dual bound to an instance: call it, or await .a(...).

ninja_devx.layers.context

Who is acting, for which tenant, in which request: passed to services without HTTP.

RequestContext dataclass

Bases: Generic[UserT, TenantT]

The acting user and tenant plus correlation ids.

Register a factory for the concrete key and inject it where needed::

container.scoped(RequestContext[User, None], request_context(User))

Outside HTTP (tasks, commands, tests) build one directly and open a scope::

with container.scope({RequestContext[User, None]: RequestContext(user, None)}) as scope:
    scope.resolve(OrderService).place(...)

user instance-attribute

user: UserT

The acting user.

tenant instance-attribute

tenant: TenantT

The acting tenant (None for single-tenant apps).

request_id class-attribute instance-attribute

request_id: str = field(default_factory=lambda: uuid4().hex)

Correlation id (X-Request-ID or generated).

trace_id class-attribute instance-attribute

trace_id: str | None = None

W3C trace id from traceparent, when present.

metadata class-attribute instance-attribute

metadata: Mapping[str, str] = field(
    default_factory=dict[str, str]
)

Free-form string metadata for logs and audit.

ninja_devx.layers.tasks

Work that must happen after the transaction commits: tasks, events, emails.

Enqueueable

Bases: Protocol[P]

django.tasks tasks (and anything similar): task.enqueue(*args, **kwargs).

TaskQueue

Bases: Protocol

Where services send background work; swap implementations in tests.

OnCommitTaskQueue dataclass

Enqueues after commit (django.tasks arguments must be JSON-serializable: pass ids).

using class-attribute instance-attribute

using: str | None = None

Database alias whose commit triggers the work.

ImmediateTaskQueue dataclass

Runs work right away (scripts, or tests that want the side effects).

RecordingTaskQueue dataclass

Records work instead of running it; run_all() executes it in order.

calls class-attribute instance-attribute

calls: list[
    tuple[object, tuple[object, ...], dict[str, object]]
] = field(
    default_factory=list[
        tuple[object, tuple[object, ...], dict[str, object]]
    ]
)

Recorded (task_or_function, args, kwargs) in order.

after_commit

after_commit(
    function: Callable[[], object],
    *,
    using: str | None = None,
) -> None

Run function when the current transaction commits, or now outside one.

ninja_devx.layers.policies

Policies: one authorization rule usable from services and (via permissions) HTTP.

Policy

Bases: Protocol[SubjectT_contra, ObjT_contra]

allows(subject, obj): the subject is usually a user or a RequestContext.

PolicyDenied

Bases: PermissionDenied

You do not have permission to perform this action.

require

require(
    policy: Policy[SubjectT, ObjT],
    subject: SubjectT,
    obj: ObjT,
) -> None

Raise PolicyDenied (HTTP 403 when mapped) unless policy allows it.

ninja_devx.layers.selectors

Selectors: read queries with a name, shared by endpoints, tasks and services.

Selector

Bases: Protocol[FilterT_contra, ModelT_co]

Returns a queryset (paginated, ordered and optimized by list endpoints) or a list.

ninja_devx.layers.persistence

Writing validated data to Django models, without any web dependency.

model_field

model_field(
    model: type[Model], name: str
) -> models.Field[object, object] | None

The concrete or many-to-many field called name (or whose attname is name).

validation_failed

validation_failed(exc: ValidationError) -> ValidationFailed

A Django ValidationError as a domain ValidationFailed (HTTP 422 when mapped).

save_instance

save_instance(
    instance: ModelT,
    data: Mapping[str, object],
    *,
    validate: bool = True,
    using: str | None = None,
) -> ModelT

Assign data to instance, validate, save and set many-to-many relations.

Foreign keys accept an instance or a primary key. On a new instance, None for a non-null field with db_default leaves the value to the database. Model.full_clean errors are raised as ValidationFailed. Runs in one transaction (a savepoint when nested).

ninja_devx.layers.testing

Fakes for testing services without a database.

RecordingTaskQueue dataclass

Records work instead of running it; run_all() executes it in order.

calls class-attribute instance-attribute

calls: list[
    tuple[object, tuple[object, ...], dict[str, object]]
] = field(
    default_factory=list[
        tuple[object, tuple[object, ...], dict[str, object]]
    ]
)

Recorded (task_or_function, args, kwargs) in order.

InMemoryRepository

Bases: Generic[ModelT]

A Repository keeping unsaved model instances in a dict (no database, no signals).

make_context

make_context(
    user: UserT, tenant: TenantT, **metadata: str
) -> RequestContext[UserT, TenantT]

A RequestContext for tests: make_context(user, None).

Parameters:

Name	Type	Description	Default
user	UserT	The acting user.	required
tenant	TenantT	The tenant, or None.	required
metadata	str	String metadata.	{}

CQRS and DDD

ninja_devx.cqrs.messages

Application messages: optional markers for commands and queries.

Subclassing is optional; any frozen dataclass works with use_case and use_query. The markers document intent and let type checkers (and devx_inspect) tell a write from a read.

Message

Base for an application message.

Command

Bases: Message

A request that changes state, handled by a use_case handler.

Query

Bases: Message

A request that reads state, handled by a use_query handler.

ninja_devx.cqrs.events

Domain events and an explicit, in-app event bus.

There is no global registry or automatic scanning: the application owns an EventBus (usually a container singleton), subscribes handlers in its wiring, and publishes events from use cases. Delivery goes through a :class:~ninja_devx.layers.TaskQueue, so handlers run after the current transaction commits.

DomainEvent

Base for a domain event; use a frozen dataclass subclass.

EventBus

A registry that delivers domain events after the current transaction commits.

Register it in the container and publish through it::

container.singleton(EventBus, EventBus(OnCommitTaskQueue()))
bus = container.resolve(EventBus)
bus.subscribe(OrderPlaced, notify_followers)
bus.publish(OrderPlaced(order_id=order.pk))

Handlers are plain callables (event) -> object. Run in-process, they enqueue django.tasks work or defer to the outbox for durable, cross-process delivery.

subscribe

subscribe(
    event: type[EventT], handler: Callable[[EventT], object]
) -> None

Call handler for every published instance of event.

Parameters:

Name	Type	Description	Default
event	type[EventT]	The event class (subclasses are not matched automatically).	required
handler	Callable[[EventT], object]	A callable receiving the event.	required

publish

publish(event: DomainEvent) -> None

Run every handler subscribed to type(event) after commit.

Parameters:

Name	Type	Description	Default
event	DomainEvent	The event instance to deliver.	required

clear

clear() -> None

Forget all subscriptions; useful when re-wiring or in tests.

ninja_devx.cqrs.unit_of_work

One transaction across several repositories: a unit of work.

ModelService already wraps each write in repository.transaction(). Use a UnitOfWork when a use case combines several repository or ORM operations that must commit or roll back together, independent of any single repository. It is synchronous; async code runs it through sync_to_async like any other transaction.atomic block.

UnitOfWork

Wrap a block in one transaction.atomic.

::

with UnitOfWork():
    orders.add(...)
    stock.change(...)

Parameters:

Name	Type	Description	Default
using	str | None	Database alias (default: Django's write router per model).	None
durable	bool	True when this must be the outermost atomic block.	False

HTTP features

ninja_devx.security.tenancy

Multi-tenancy: find the current tenant once per request and scope queries with it.

Model controllers with tenant_field filter every query by the tenant, set it on create, reject input schemas that accept it, and 404 for other tenants' objects::

class ProjectController(CRUDController[Project, ProjectOut, ProjectIn]):
    tenant_field = "organization"

The tenant comes from the first source that is configured:

1. tenant_resolver on the controller, then NINJA_DEVX["TENANT_RESOLVER"]: a function of the request (sync or async def) returning the tenant or None;
2. tenant_context on the controller, then NINJA_DEVX["TENANT_CONTEXT"]: a RequestContext[User, Tenant] key resolved from the controller's container;
3. request.tenant, as set by tenant middleware (django-tenants and similar).

A request without a tenant gets 403. The tenant is cached on the request, so services can read it with current_tenant(request).

MissingTenant

Bases: PermissionDenied

No tenant is associated with this request.

current_tenant

current_tenant(request: HttpRequest) -> object | None

The tenant already resolved for this request, or None.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The current request.	required

current_tenant_for

current_tenant_for(
    request: HttpRequest,
    *,
    resolver: TenantResolver | None,
    context: object | None,
    invocation: Invocation | None,
) -> object

Resolve (once) and return the tenant; raises MissingTenant (403) without one.

acurrent_tenant_for async

acurrent_tenant_for(
    request: HttpRequest,
    *,
    resolver: TenantResolver | None,
    context: object | None,
    invocation: Invocation | None,
) -> object

current_tenant_for for async operations: sync resolvers run in one thread hop.

ninja_devx.http.conditional

Conditional requests (RFC 9110): ETag, If-None-Match → 304, If-Match → 412.

Model controllers enable them with etag::

class PostController(CRUDController[Post, PostOut, PostIn]):
    etag = ETag()                        # tag = hash of the rendered representation
    etag = ETag(field="updated_at")      # tag = the version field: no extra serialization
    etag = ETag(require_if_match=True)   # writes without If-Match get 428

• GET /{pk} sends ETag and answers If-None-Match with 304 before serializing.
• GET / sends an ETag of the rendered page and answers If-None-Match with 304.
• PUT/PATCH/DELETE compare If-Match with the current object and fail with 412 when it changed in between (optimistic locking); successful writes send the new tag.

Any other operation can use the response-level version: decorators=[conditional()]. Weak comparison is used for If-None-Match (RFC 9110 §13.1.2); If-Match: * always passes.

PreconditionFailed

Bases: DomainError

The resource changed since you last fetched it.

PreconditionRequired

Bases: DomainError

Send If-Match with the ETag you last received.

ETag dataclass

field class-attribute instance-attribute

field: str | None = None

A field that changes on every write (updated_at, version). None hashes the output schema's representation instead, which costs one extra serialization per check.

weak class-attribute instance-attribute

weak: bool = False

Weak tags (W/"...") survive compression and JSON formatting differences.

require_if_match class-attribute instance-attribute

require_if_match: bool = False

Writes without If-Match fail with 428 Precondition Required.

lists class-attribute instance-attribute

lists: bool = True

Also tag list responses (from their rendered body).

entity_tag

entity_tag(value: object, *, weak: bool = True) -> str

A quoted entity tag for value (strings, numbers, dates, JSON-like data).

remember_etag

remember_etag(request: HttpRequest, tag: str) -> None

Send tag as the response's ETag (used by conditional()-wrapped views).

not_modified

not_modified(
    request: HttpRequest, tag: str
) -> HttpResponseNotModified | None

A 304 when If-None-Match matches tag on a safe request.

check_if_match

check_if_match(
    request: HttpRequest, tag: str, *, required: bool
) -> None

412 when If-Match does not match tag; 428 when required and missing.

conditional

conditional(*, from_body: bool = True) -> Callable[[F], F]

An operation decorator adding ETag and 304 handling to successful GET responses.

With from_body the tag hashes the rendered response; otherwise only tags set with remember_etag are used.

Parameters:

Name	Type	Description	Default
from_body	bool	Hash the rendered body when no tag was set with remember_etag.	True

ninja_devx.http.throttling

Rate limits for Ninja's throttle= option: per user, per client, per tenant, per scope.

::

NINJA_DEVX = {"THROTTLE_RATES": {"uploads": "10/min", "search": "120/min"}}

class UploadController(Controller):
    options = ControllerOptions(throttle=[UserRateThrottle("1000/day")])

    @post("/", throttle=[ScopedRateThrottle("uploads"), ClientRateThrottle(anon="5/min")])
    def upload(self, request: HttpRequest, file: UploadedFile) -> Upload: ...

They plug into Ninja's own throttling, so rejected requests get Ninja's 429 with a Retry-After header. Unlike Ninja's built-ins they:

• keep no per-request state on the (shared) throttle object, so they are thread-safe;
• count in fixed windows through a ThrottleStorage, so limits hold across processes;
• read NINJA_DEVX["THROTTLE_RATES"] at request time, so override_settings works;
• accept rates like "100/min", "1000/day" or "20/5min".

The default storage counts with cache.add + cache.incr, which is atomic on Redis and Memcached but not elsewhere (a crash between the two calls loses one hit). Pass storage= (or set NINJA_DEVX["THROTTLE_STORAGE"]) to plug in something stricter, such as ninja_devx.contrib.redis_throttle.RedisThrottleStorage.

ThrottleStorage

Bases: Protocol

Where throttles count hits. key already identifies the current fixed window.

hit

hit(key: str, window_seconds: int) -> tuple[int, float]

Count one hit against key, creating its window (window_seconds TTL) on the first hit, and return (count, retry_after): the count including this hit, and the seconds remaining until the window resets.

Parameters:

Name	Type	Description	Default
key	str	Unique to the throttle, identity and current window.	required
window_seconds	int	The window's length, used as the new key's TTL.	required

CacheThrottleStorage dataclass

Default storage: a Django cache, via cache.add + cache.incr.

Atomic on Redis and Memcached, so limits hold across processes; on other backends a crash between the two calls can lose one hit.

cache_alias class-attribute instance-attribute

cache_alias: str = 'default'

Cache alias (settings.CACHES) counters are stored under.

RateThrottle

Bases: BaseThrottle

Base class: rate requests per window for each identity from identify().

identify

identify(request: HttpRequest) -> str | None

The identity to count for, or None to not throttle this request.

UserRateThrottle

Bases: RateThrottle

Per authenticated user (request.auth or request.user); anonymous requests pass.

AnonRateThrottle

Bases: RateThrottle

Per client IP for anonymous requests; authenticated requests pass.

ClientRateThrottle

Bases: RateThrottle

One throttle, two rates: ClientRateThrottle(user="600/min", anon="30/min").

ScopedRateThrottle

Bases: ClientRateThrottle

A named limit whose rate comes from NINJA_DEVX["THROTTLE_RATES"][scope].

Each user (or anonymous client IP) has its own counter per scope, so ScopedRateThrottle("uploads") on several operations shares one budget.

TenantRateThrottle

Bases: RateThrottle

Per tenant: the tenant already resolved, request.tenant, or tenant(request).

parse_rate

parse_rate(rate: str) -> tuple[int, int]

"100/min" → (100, 60); "20/5min" → (20, 300).

Parameters:

Name	Type	Description	Default
rate	str	Requests per period, e.g. "100/min" or "20/5min".	required

ninja_devx.serialization.visibility

Field visibility by role: one output schema, fields shown only to who may see them.

::

class EmployeeOut(FieldVisibility, Schema):
    id: int
    name: str
    salary: Annotated[Decimal | None, VisibleTo(IsStaff())] = None
    email: Annotated[str | None, VisibleTo(IsStaff() | IsOwner("user"), hidden="omit")] = None

• VisibleTo takes the same permissions as operations (&, |, ~ work).
• A hidden field is serialized as null (hidden="null", the default) or left out (hidden="omit"). Declare it optional (T | None = None) so OpenAPI and clients know.
• Request-level checks (IsStaff, HasDjangoPermission) work on any Schema.
• Object-level checks (IsOwner, a Policy) and hidden="omit" need the FieldVisibility mixin, which knows the object being serialized. Without it an object-level rule hides the field: it fails closed.
• Checks run during serialization, with the request Ninja passes to pydantic, so they are synchronous; in async operations the user is loaded before (aprepare_request).

VisibleTo dataclass

__init__

__init__(
    *permissions: BasePermission[Never],
    hidden: Literal["null", "omit"] = "null",
) -> None

Parameters:

Name	Type	Description	Default
permissions	BasePermission[Never]	All must allow; each counts as its request check and its object check.	()
hidden	Literal['null', 'omit']	"null" serializes a hidden field as null; "omit" leaves it out (needs FieldVisibility).	'null'

WriteVisibleTo dataclass

A field only some callers may write; checked by model controllers before persisting.

Use request-level permissions (IsStaff, a policy on request); object-level checks fail closed because the object is not known when the payload arrives::

class ArticleIn(Schema):
    title: str
    featured: Annotated[bool, WriteVisibleTo(IsStaff())] = False

A non-staff create or update that sends featured is rejected with 403.

__init__

__init__(*permissions: BasePermission[Never]) -> None

Parameters:

Name	Type	Description	Default
permissions	BasePermission[Never]	All must allow the request, or the field is rejected.	()

Expandable dataclass

A relation rendered as its key, or as a nested schema when the client asks.

author: Annotated[int | UserOut, Expandable()] is 1 by default and {"id": 1, "username": "ada"} with ?expand=author. The unexpanded form reads only the foreign key column (author_id), and the query planner joins the relation only when it is expanded. Needs the FieldVisibility mixin on the schema.

source class-attribute instance-attribute

source: str | None = None

Attribute holding the key when not expanded (default <field>_id).

ResponseShape dataclass

What the client asked for (?fields= and ?expand=) for one response schema.

FieldVisibility

Schema mixin for response shaping: object-level VisibleTo rules, hidden="omit", ?fields= sparse fieldsets and Expandable relations.

List it before Schema: class EmployeeOut(FieldVisibility, Schema).

write_markers

write_markers(
    schema: type[object],
) -> dict[str, WriteVisibleTo]

{field name: WriteVisibleTo} of an input schema.

forbidden_writes

forbidden_writes(
    schema: type[object],
    sent: object,
    request: HttpRequest | None,
) -> set[str]

Names of sent fields the caller may not write for schema.

Parameters:

Name	Type	Description	Default
schema	type[object]	The input schema.	required
sent	object	Field names actually submitted.	required
request	HttpRequest | None	The current request.	required

expandable_fields

expandable_fields(
    schema: type[object],
) -> dict[str, Expandable]

{field name: Expandable} of a pydantic schema.

ninja_devx.serialization.privacy

Mark schema fields that must never appear in logs, error echoes or plain exports.

::

class UserIn(Schema):
    email: str
    password: Annotated[str, Sensitive()]

Sensitive is metadata only; it does not change validation or serialization by itself. Callers that might otherwise echo raw field values (the CRUD CSV/JSONL export, a custom error body) call redact_payload on the data they are about to emit.

Sensitive dataclass

Field metadata: Annotated[str, Sensitive()]. Carries no configuration.

sensitive_fields

sensitive_fields(schema: type[object]) -> frozenset[str]

Field and alias names on schema annotated with Sensitive.

Both spellings are returned because a dump may be keyed by either, depending on whether it used by_alias.

Parameters:

Name	Type	Description	Default
schema	type[object]	A pydantic model (or Schema).	required

mask

mask(value: object) -> object

Replace a sensitive value with "***" (None stays None; lists are masked element-wise).

Parameters:

Name	Type	Description	Default
value	object	The value to mask.	required

redact_payload

redact_payload(
    schema: type[object], data: Mapping[str, object]
) -> dict[str, object]

A copy of data with every Sensitive field of schema masked.

Parameters:

Name	Type	Description	Default
schema	type[object]	The pydantic model data was (or will be) dumped from.	required
data	Mapping[str, object]	A mapping keyed by field name or alias, such as a model_dump() result.	required

ninja_devx.serialization.examples

OpenAPI examples generated from ninja_devx.testing.sample.

::

ArticleOut = with_examples(ArticleOut)

Built field by field, so one field with no sample rule (no default, no example, no type ninja_devx.testing.sample knows how to derive) is just left out instead of failing the whole schema. Model controllers do this automatically for their input and output schemas with openapi_examples = True (see ninja_devx.crud.controllers.ModelController).

with_examples

with_examples(schema: type[SchemaT]) -> type[SchemaT]

schema, with an OpenAPI example filled in from its fields' derived sample values.

Mutates schema.model_config in place and returns the same class, so it composes with the rest of a schema's definition: ArticleOut = with_examples(ArticleOut).

Parameters:

Name	Type	Description	Default
schema	type[SchemaT]	The schema to add an example to.	required

ninja_devx.http.middleware

Middleware for Ninja routers and APIs, including plain function views.

Built on Ninja's add_decorator(..., mode="view"): a middleware wraps the whole operation (authentication, throttling, validation and the view), so it also sees the 401/422/429 responses::

use_middleware(api, RequestIDMiddleware(), ServerTimingMiddleware())   # every operation
use_middleware(router, DeprecationMiddleware(sunset=datetime(2027, 1, 1, tzinfo=UTC)))
ControllerOptions(middleware=[RateLimitHeadersMiddleware()])            # one controller
mount(api, V1, prefix="/v1", middleware=[DeprecationMiddleware(...)])   # a version

Write your own by subclassing Middleware and overriding process_request (return a response to short-circuit) and/or process_response. Override the a-prefixed methods too when the work is I/O in async operations.

Middleware

Hooks around an operation. Both methods are optional.

process_request

process_request(
    request: HttpRequest,
) -> HttpResponseBase | None

Runs first; return a response to skip the operation.

process_response

process_response(
    request: HttpRequest, response: HttpResponseBase
) -> HttpResponseBase

Runs last, on every response (including errors).

process_exception

process_exception(
    request: HttpRequest, exception: BaseException
) -> None

Release per-request resources when request, handler or response processing fails.

This is a cleanup hook; the original exception (including cancellation) is re-raised.

RequestIDMiddleware dataclass

Bases: Middleware

Accept or create a request id, expose it to the request and echo it in the response.

header class-attribute instance-attribute

header: str = 'X-Request-ID'

Header read from the request and written to the response.

generate class-attribute instance-attribute

generate: Callable[[], str] = field(
    default=lambda: uuid.uuid4().hex
)

Makes an id when the client sends none.

ServerTimingMiddleware

Bases: Middleware

Server-Timing: app;dur=<ms> for browser dev tools and APM.

DeprecationMiddleware dataclass

Bases: Middleware

Deprecation (RFC 9745), Sunset (RFC 8594) and Link headers for old versions.

deprecated_at class-attribute instance-attribute

deprecated_at: datetime | None = None

When the API was deprecated; None sends Deprecation: true.

sunset class-attribute instance-attribute

sunset: datetime | None = None

When it stops working.

link class-attribute instance-attribute

link: str | None = None

Migration guide URL (Link: <...>; rel="deprecation").

RateLimitHeadersMiddleware

Bases: Middleware

RateLimit-Limit/Remaining/Reset and RateLimit-Policy from ninja-devx throttles.

The most restrictive throttle of the request is reported. Added automatically to controllers whose throttle uses ninja_devx.http.throttling classes.

get_request_id

get_request_id(request: HttpRequest) -> str | None

The id RequestIDMiddleware accepted or created for request.

Parameters:

Name	Type	Description	Default
request	HttpRequest	The current request.	required

middleware_decorator

middleware_decorator(
    *middlewares: Middleware,
) -> Callable[[Run], Run]

A Ninja mode="view" decorator running middlewares (first is outermost).

Parameters:

Name	Type	Description	Default
middlewares	Middleware	Middleware instances; the first one is outermost.	()

use_middleware

use_middleware(
    target: NinjaAPI | Router, *middlewares: Middleware
) -> None

Apply middlewares to every operation of an API or router (call before mounting).

Parameters:

Name	Type	Description	Default
target	NinjaAPI | Router	A NinjaAPI or a Router.	required
middlewares	Middleware	Middleware instances; the first one is outermost.	()

record_rate_limit

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).

Parameters:

Name	Type	Description	Default
request	HttpRequest	The current request.	required
policy	str	Throttle name, shown in RateLimit-Policy.	required
limit	int	Requests allowed per window.	required
remaining	int	Requests left in the current window.	required
reset	float	Seconds until the window resets.	required
window	int	Window length in seconds.	required

ninja_devx.http.versioning

Serve an older response shape, negotiated by a request header.

Declare the schemas a controller used to return, keyed by the version number clients asked for; the current shape is whatever the operation already declares with response=::

class PostOutV1(Schema):
    id: int
    title: str

class PostController(VersionedResponseMixin, CRUDController[Post, PostOut, PostIn]):
    response_versions = {1: PostOutV1}

A client sending Accept-Version: 1 gets PostOutV1 (and list[PostOutV1] for list operations), re-validated from the very same rendered response, so there is no second database round trip. Omitting the header serves the latest shape. Every response carries X-API-Version naming the version actually served; an unknown Accept-Version answers 406 in the project's error format.

Only single-schema and list[...] responses can be downgraded; other shapes (status-keyed responses without one 2xx entry, dict bodies...) still get the header and 406 handling, but are served unchanged since there is no schema to validate them against.

VersionedResponseMiddleware

Bases: Middleware

Negotiate Accept-Version and downgrade the rendered response.

Added automatically by VersionedResponseMixin when response_versions is set; construct it directly only to use it without the mixin.

Parameters:

Name	Type	Description	Default
response_versions	Mapping[int, type[BaseModel]]	{version: schema} for every version but the latest.	required
header	str	Request header naming the wanted version.	'Accept-Version'
response_header	str	Response header naming the version actually served.	'X-API-Version'

VersionedResponseMixin

Bases: Controller

Adds Accept-Version negotiation to every operation of a controller.

List it first so its merged_options/customize_operation see the final response=: class PostController(VersionedResponseMixin, CRUDController[...]).

response_versions class-attribute

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 class-attribute

response_version_header: str = 'Accept-Version'

Request header naming the wanted version.

response_version_response_header class-attribute

response_version_response_header: str = 'X-API-Version'

Response header naming the version actually served.

ninja_devx.http.explain

X-Query-Count, X-Query-Time and X-Query-Plan headers, for development.

Install it where a paginated or optimized controller is mounted::

use_middleware(api, QueryExplainMiddleware())             # active only when DEBUG
use_middleware(api, QueryExplainMiddleware(enabled=True))  # a staff-only diagnostics API

Counts and durations come from django.db.connection.queries (CaptureQueriesContext semantics: force_debug_cursor is set for the request, so this works even when DEBUG is off). X-Query-Plan lists the select_related/prefetch_related lookups the N+1 planner chose, read from the request attribute it records them on (QUERY_PLAN_ATTR, filled the same way pagination metadata is). SQL text never reaches a header, and every hook is a no-op when the middleware is inactive, so nothing is captured in production by default.

Database connections are thread-local: an async list evaluated off-thread (run_sync, or the plain sync_to_async fallback for a queryset without an async paginator) runs its queries against a connection this middleware never touched, so the count can undercount for async operations whose ORM access hops threads.

QUERY_PLAN_ATTR module-attribute

QUERY_PLAN_ATTR: Final = '_ninja_devx_query_plan'

Request attribute the N+1 planner fills with (select_related, prefetch_related) lookup names it chose for the current request.

QueryExplainMiddleware

Bases: Middleware

Add query diagnostics headers to matching responses.

Parameters:

Name	Type	Description	Default
enabled	bool | None	Forces the middleware on or off; None (the default) follows settings.DEBUG on every request, so override_settings(DEBUG=...) works.	None
databases	Sequence[str]	Database aliases to count; defaults to every configured alias.	()

ninja_devx.http.requestlog

One structured log record per request: identity, timing and outcome, never the body.

::

install(api, [RequestLogPlugin()])

LOGGING = {
    "version": 1,
    "formatters": {"json": {"()": "ninja_devx.http.requestlog.JSONFormatter"}},
    "handlers": {"console": {"class": "logging.StreamHandler", "formatter": "json"}},
    "loggers": {"ninja_devx.request": {"handlers": ["console"], "level": "INFO"}},
}

Fields: request_id, method, path, status, duration_ms, user_id, tenant (the resolved tenant, or request.tenant as tenant middleware sets it; None without one), query_count (queries on the default database connection), operation ("Controller.method"), and api_key_prefix when ninja_devx.contrib.apikeys authenticated the request. naming="otel" (the default) spells the four fields with an OpenTelemetry semantic convention name (http.request.method, url.path, http.response.status_code, user.id); naming="flat" keeps the plain names above.

When structlog <https://www.structlog.org/>_ is installed, the identity fields (request_id, method, path, user id, tenant) are also bound with structlog.contextvars.bind_contextvars as soon as the request starts and cleared once the final record is logged, so the application's own structlog calls carry them too; fields that are only known once the handler returns (status, duration, query count, operation) are not part of that binding.

RequestLogMiddleware dataclass

Bases: Middleware

Logs one record per request. Install directly, or through RequestLogPlugin.

logger class-attribute instance-attribute

logger: Logger = field(
    default_factory=lambda: logging.getLogger(
        "ninja_devx.request"
    )
)

Logger receiving one record per request.

level class-attribute instance-attribute

level: int = logging.INFO

Log level of the record.

naming class-attribute instance-attribute

naming: Naming = 'otel'

"otel": OpenTelemetry semantic convention field names; "flat": plain ones.

bind_structlog class-attribute instance-attribute

bind_structlog: bool = True

Bind the identity fields to structlog.contextvars when structlog is installed.

RequestLogPlugin dataclass

Bases: APIPlugin

RequestLogMiddleware plus RequestIDMiddleware, so every request gets an id.

::

install(api, [RequestLogPlugin(naming="flat")])

logger class-attribute instance-attribute

logger: Logger = field(
    default_factory=lambda: logging.getLogger(
        "ninja_devx.request"
    )
)

Logger receiving one record per request.

level class-attribute instance-attribute

level: int = logging.INFO

Log level of the record.

naming class-attribute instance-attribute

naming: Naming = 'otel'

"otel": OpenTelemetry semantic convention field names; "flat": plain ones.

bind_structlog class-attribute instance-attribute

bind_structlog: bool = True

Bind the identity fields to structlog.contextvars when structlog is installed.

include_request_id class-attribute instance-attribute

include_request_id: bool = True

Add RequestIDMiddleware too, so request_id is never empty.

JSONFormatter

Bases: Formatter

Render each LogRecord as one JSON object, extra fields included.

::

"formatters": {"json": {"()": "ninja_devx.http.requestlog.JSONFormatter"}}

register_api_key_reader

register_api_key_reader(reader: ApiKeyReader) -> None

Let ninja_devx.contrib.apikeys (or your own auth) contribute api_key_prefix.

Apps register in AppConfig.ready; core never imports contrib.

Parameters:

Name	Type	Description	Default
reader	ApiKeyReader	Returns the authenticating key's prefix, or None.	required

ninja_devx.http.security

Security response headers as a middleware.

Applies conservative, API-friendly defaults and never overwrites a header the view already set. HSTS is opt-in because it requires HTTPS::

use_middleware(api, SecurityHeadersMiddleware(hsts="max-age=31536000; includeSubDomains"))

SecurityHeadersMiddleware

Bases: Middleware

Set security headers on every response unless the response already has them.

Parameters:

Name	Type	Description	Default
hsts	str | None	Strict-Transport-Security value, or None to leave it unset.	None
csp	str | None	Content-Security-Policy value, or None to leave it unset.	None
referrer_policy	str	Referrer-Policy value.	'same-origin'
frame_options	str | None	X-Frame-Options value; None leaves it unset.	'DENY'
permissions_policy	str | None	Permissions-Policy value; None leaves it unset.	_PERMISSIONS_POLICY
nosniff	bool	set X-Content-Type-Options: nosniff.	True
extra	Mapping[str, str] | None	additional headers, applied after the defaults.	None

ninja_devx.http.cache

Cache whole HTTP responses for read operations.

ResponseCacheMiddleware caches the body, status and content type of matching responses and adds Cache-Control/Vary. Install it on an API, router or controller::

use_middleware(api, ResponseCacheMiddleware(ttl=60, vary_on=("Authorization",)))

Invalidate every cached response sharing a prefix with invalidate_cache after a write. Cache calls are synchronous, also for async operations.

ResponseCacheMiddleware

Bases: Middleware

Serve matching responses from the cache and store new ones.

Cache-Control is private when vary_on names Authorization or Cookie and public otherwise, unless the response already carries the header.

Parameters:

Name	Type	Description	Default
ttl	int	Cache lifetime in seconds.	60
vary_on	Sequence[str]	Header names that participate in the cache key (for example ("Authorization", "Accept-Language")).	()
cache	str	Cache alias.	'default'
key_prefix	str	Prefix used by invalidate_cache.	''
methods	Sequence[str]	HTTP methods to cache (others pass through).	('GET', 'HEAD')

invalidate_cache

invalidate_cache(
    prefix: str = "", *, cache: str = "default"
) -> None

Invalidate every cached response under prefix.

Parameters:

Name	Type	Description	Default
prefix	str	The same prefix passed to ResponseCacheMiddleware.	''
cache	str	Cache alias.	'default'

ninja_devx.http.hardening

Request hardening: body size, content type and JSON nesting limits.

Each is a :class:~ninja_devx.http.middleware.Middleware returning a short-circuiting response in the package's error format::

use_middleware(
    api,
    MaxBodySizeMiddleware(5_000_000),
    EnforceContentTypeMiddleware({"application/json"}),
    JsonDepthMiddleware(max_depth=32),
)

MaxBodySizeMiddleware

Bases: Middleware

Reject requests whose Content-Length exceeds max_bytes with 413.

Only the declared length is checked; chunked bodies without one are bounded by Django's DATA_UPLOAD_MAX_MEMORY_SIZE.

Parameters:

Name	Type	Description	Default
max_bytes	int	Maximum accepted body size in bytes.	required

EnforceContentTypeMiddleware

Bases: Middleware

Reject write requests whose media type is not allowed, with 415.

A request without a Content-Type header passes.

Parameters:

Name	Type	Description	Default
media_types	Collection[str]	Accepted media types (parameters such as ; charset= ignored).	required
methods	Collection[str]	Methods to check (default: POST/PUT/PATCH/QUERY).	_BODY_METHODS

JsonDepthMiddleware

Bases: Middleware

Reject JSON bodies nested deeper than max_depth with 400.

Parameters:

Name	Type	Description	Default
max_depth	int	Maximum nesting depth of objects and arrays.	32

ninja_devx.http.pagination_headers

Expose pagination metadata as response headers (RFC 8288 Link, X-Total-Count).

Add it where the paginated controller is mounted::

use_middleware(api, PaginationHeadersMiddleware())

Link carries rel="next"/rel="prev" for both paginators; X-Total-Count is set when limit/offset pagination computed a count. A count option that only counts up to a threshold sends X-Total-Count: N+ past it, instead of N.

PAGINATION_ATTR module-attribute

PAGINATION_ATTR: Final = '_ninja_devx_pagination'

Request attribute a paginator fills with count/next/previous.

PaginationHeadersMiddleware

Bases: Middleware

Copy pagination metadata recorded by the paginator onto response headers.

ninja_devx.http.health

Liveness and readiness endpoints for load balancers and orchestrators.

::

mount(api, {"/health": HealthController})          # GET /health/live, GET /health/ready

class Health(HealthController):
    health_checks = (DatabaseCheck(), CacheCheck(), MigrationsCheck(), MyQueueCheck())

/live only proves the process answers. /ready runs every check and returns 200 or 503 with the failing checks, so traffic stops before users see errors. Neither requires authentication.

HealthCheck

Bases: Protocol

name property

name: str

Shown in the report.

check

check() -> None

Raise when the dependency is not usable.

DatabaseCheck dataclass

Runs SELECT 1 on a database connection.

CacheCheck dataclass

Writes and reads a key in a cache.

MigrationsCheck dataclass

Fails while migrations are not applied (useful during deploys).

HealthController

Bases: Controller

GET /live and GET /ready.

health_checks class-attribute

health_checks: Sequence[HealthCheck] = (DatabaseCheck(),)

Checks run concurrently by /ready; results retain this order.

health_timeout class-attribute

health_timeout: float = 2.0

Total readiness deadline in seconds. Hung checks retain their bounded worker slot.

ninja_devx.serialization.renderers

Faster JSON renderers for Ninja: NinjaAPI(renderer=ORJSONRenderer()).

Serialization of the response body is often the largest per-request cost after the database. Both renderers accept what Ninja's JSONRenderer does (datetimes, UUIDs, Decimal, pydantic models) and produce the same JSON, except that datetimes keep their microseconds (Django's encoder truncates them to milliseconds).

• ORJSONRenderer needs orjson (pip install ninja-devx[orjson]).
• MsgspecRenderer needs msgspec (pip install ninja-devx[msgspec]).

ORJSONRenderer

Bases: BaseRenderer

JSON with orjson (ninja-devx[orjson]): NinjaAPI(renderer=ORJSONRenderer()).

Output matches Ninja's encoder (Z datetimes, decimals as strings) except that datetimes keep microseconds.

options class-attribute

options: int | None = None

Extra orjson.OPT_* flags (orjson.OPT_INDENT_2...).

MsgspecRenderer

Bases: BaseRenderer

JSON with msgspec (ninja-devx[msgspec]); same output rules as ORJSONRenderer.

Contrib apps

ninja_devx.contrib.apikeys.auth

Creating, authenticating and scoping API keys.

KEEP module-attribute

KEEP: Final = object()

Sentinel: keep the current value when rotating a key.

APIKeyAuth

Bases: APIKeyHeader

X-API-Key: ndx_...; request.auth is the key's user.

In async operations Ninja runs it in one thread hop (lookup and last_used_at together), which is cheaper than two async ORM calls.

APIKeyBearer

Bases: HttpBearer

Authorization: Bearer ndx_....

RequiresScope dataclass

Operation metadata: meta=(RequiresScope("orders:write"),).

HasScopes dataclass

Bases: BasePermission[object]

Every RequiresScope of the operation must be granted to the request's API key.

Requests authenticated otherwise (sessions, JWT) pass when allow_unscoped is set, since their scopes are the user's permissions.

allow_unscoped class-attribute instance-attribute

allow_unscoped: bool = False

Let requests without an API key through (e.g. session users of the same API).

APIKeyRateThrottle

Bases: RateThrottle

Per API key: the key's own rate_limit, else rate; other requests pass.

ControllerOptions(throttle=[APIKeyRateThrottle("1000/hour"), UserRateThrottle(...)])

create_api_key

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.

Parameters:

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

revoke_api_key

revoke_api_key(key: APIKey) -> None

Stop accepting key (kept for audit).

Parameters:

Name	Type	Description	Default
key	APIKey	The key to revoke.	required

rotate_api_key

rotate_api_key(
    key: APIKey,
    *,
    scopes: Iterable[str] | None = None,
    expires_at: object = KEEP,
    rate_limit: object = KEEP,
) -> tuple[APIKey, str]

Replace key's secret in place and return the new raw value.

The old secret stops working immediately. The row (owner, name, creation time) is kept; pass scopes to update them, and expires_at/rate_limit (including None to clear them) to change the expiry or the limit. A revoked key stays revoked: create a new one instead.

Parameters:

Name	Type	Description	Default
key	APIKey	The key to rotate.	required
scopes	Iterable[str] | None	New scopes, or None to keep the current ones.	None
expires_at	object	New expiry; omit to keep the current one.	KEEP
rate_limit	object	New rate limit; omit to keep the current one.	KEEP

Raises:

Type	Description
ValueError	key is revoked.

current_api_key

current_api_key(request: HttpRequest) -> APIKey | None

The API key that authenticated request, if any.

scope_allows

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

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

Parameters:

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

ninja_devx.contrib.apikeys.api

Self-service key management for the authenticated user.

APIKeyController

Bases: Controller

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

grantable_scopes class-attribute

grantable_scopes: Sequence[str] | None = None

Scopes users may put on their keys (None: any).

ninja_devx.contrib.grants.backends

Object-permission backends backed by ObjectGrant.

GrantBackend is a Django authentication backend answering object-level has_perm. GrantsBackend implements the ninja_devx.security.object_permissions backend protocol and is registered by GrantsConfig.ready so core does not import this app.

GrantBackend

Add to AUTHENTICATION_BACKENDS next to ModelBackend; it never authenticates.

GrantsBackend

Object permission backend stored in ninja_devx.contrib.grants.ObjectGrant.

object_permissions_of

object_permissions_of(user: object, obj: Model) -> set[str]

{"app_label.codename", ...} granted to user (directly or via groups) on obj.

Cached on the user object for its lifetime (usually one request), like Django's own permission caches.

ninja_devx.contrib.audit.log

Recording audit entries and diffing model snapshots.

AuditMixin

Bases: ModelController[ModelT], Generic[ModelT]

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

List it first: class Invoices(AuditMixin[Invoice], CRUDController[...]). Your own perform_* overrides on the controller are audited as well.

audit_privacy class-attribute

audit_privacy: AuditPrivacy = AuditPrivacy()

Storage allowlists, credential redaction and safe object labels.

audit_fields class-attribute

audit_fields: Sequence[str] | None = None

Fields kept in snapshots and diffs (None: every concrete field).

audit_exclude class-attribute

audit_exclude: Sequence[str] = ('password',)

Fields never recorded.

audit_redact class-attribute

audit_redact: Sequence[str] = ()

Fields recorded as "***": a change is visible, the value is not.

audit_metadata

audit_metadata(
    request: HttpRequest,
) -> Mapping[str, object] | None

Extra data stored with every entry (override: {"tenant": ...}).

audit

audit(
    request: HttpRequest,
    action: str,
    instance: ModelT,
    *,
    changes: Mapping[str, list[object]],
    object_pk: object = None,
) -> AuditEntry

Write one entry; override to enrich or filter what is recorded.

snapshot

snapshot(
    instance: 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).

Parameters:

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

diff

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

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

Parameters:

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

redact

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

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

Parameters:

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

record

record(
    request: HttpRequest | None,
    action: str,
    obj: 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.

Parameters:

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

ninja_devx.contrib.audit.api

Read-only audit log endpoints.

AuditFilters

Bases: FilterSchema

Query parameters of GET / on AuditLogController.

action class-attribute instance-attribute

action: str | None = None

create, update, delete or a custom action.

actor_id class-attribute instance-attribute

actor_id: int | str | UUID | None = None

Entries by this user.

model class-attribute instance-attribute

model: str | None = Field(
    None, description="`app_label.model`."
)

Entries about this model: app_label.model.

object_pk class-attribute instance-attribute

object_pk: str | None = None

Entries about this object (with model).

request_id class-attribute instance-attribute

request_id: str | None = None

Everything recorded during one request.

since class-attribute instance-attribute

since: Annotated[
    datetime | None, FilterLookup("created__gte")
] = None

Entries at or after this time.

until class-attribute instance-attribute

until: Annotated[
    datetime | None, FilterLookup("created__lt")
] = None

Entries before this time.

AuditLogController

Bases: ReadOnlyModelController[AuditEntry, AuditEntryOut]

GET / (filterable) and GET /{pk}; staff only by default.

AuditHistoryMixin

Bases: ModelController[ModelT], Generic[ModelT]

GET /{pk}/history: the object's audit entries, newest first.

The object is loaded like retrieve (scoping, object permissions: 404/403), so staff access is additionally required by default. An explicit routes["history"] permission override can implement a project-specific audit-reader policy.

ninja_devx.contrib.audit.privacy

Explicit storage policy for audit changes and metadata.

AuditPrivacy dataclass

Allowlist business fields and redact credential values before persistence.

Field names are exact; metadata credential keys are matched case-insensitively. Use an allowlist for models containing personal or application-specific secrets.

fields class-attribute instance-attribute

fields: tuple[str, ...] | None = None

Allowlist of change field names to keep (None: keep every field not excluded).

exclude class-attribute instance-attribute

exclude: tuple[str, ...] = (
    "password",
    "hashed_secret",
    "private_key",
)

Change field names dropped entirely, never stored even redacted.

redact class-attribute instance-attribute

redact: tuple[str, ...] = (
    "secret",
    "token",
    "access_token",
    "refresh_token",
    "api_key",
    "authorization",
)

Change field names stored as "***" instead of their value.

metadata_fields class-attribute instance-attribute

metadata_fields: tuple[str, ...] | None = None

Allowlist of metadata keys to keep (None: keep every key).

object_repr class-attribute instance-attribute

object_repr: bool = False

Opt in to calling the model's potentially sensitive __str__.

schema class-attribute instance-attribute

schema: type[object] | None = None

A schema (ninja_devx.serialization.privacy.Sensitive-annotated) whose marked fields are redacted like redact names.

ninja_devx.contrib.webhooks.outbox

Publishing events and delivering them to endpoints.

RETRY_SCHEDULE module-attribute

RETRY_SCHEDULE: Final = (
    timedelta(seconds=5),
    timedelta(minutes=5),
    timedelta(minutes=30),
    timedelta(hours=2),
    timedelta(hours=5),
    timedelta(hours=10),
    timedelta(hours=10),
)

Delays before the retries (Standard Webhooks' schedule): 8 attempts over ~27 hours.

Transport

Bases: Protocol

Sends one request; returns the status code or raises OSError.

event_matches

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

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

Parameters:

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

publish

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).

Without queue a worker (devx_webhooks deliver) sends it. With a queue, delivery is also enqueued right away, after commit with OnCommitTaskQueue; the worker still retries failures.

Parameters:

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

deliver_event

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

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

Parameters:

Name	Type	Description	Default
event_id	str	The OutboxEvent primary key.	required
using	str | None	Database alias containing the outbox event.	None

body_of

body_of(event: OutboxEvent) -> bytes

The JSON request body: {"type", "timestamp", "data"}.

deliver_due

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).

Parameters:

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

ninja_devx.contrib.webhooks.signing

Standard Webhooks signatures (https://www.standardwebhooks.com).

InvalidSignature

Bases: Exception

The webhook request is not authentic, or too old.

generate_secret

generate_secret() -> str

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

sign

sign(
    secret: str,
    message_id: str,
    timestamp: int,
    body: bytes,
) -> str

v1,<base64 HMAC-SHA256 of "id.timestamp.body">.

signature_headers

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.

Parameters:

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

verify_signature

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_.

For receivers (including Django views: verify_signature(secret, request.headers, request.body)). Several secrets allow rotation; tolerance (seconds) rejects replayed old requests.

Parameters:

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

ninja_devx.contrib.webhooks.api

Managing webhook endpoints over the API.

WebhookEndpointController

Bases: CRUDController[WebhookEndpoint, WebhookEndpointOut, WebhookEndpointIn]

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

url_policy class-attribute

url_policy: URLPolicy = URLPolicy()

Which URLs may be registered (HTTPS, public addresses by default). Pass the same policy to the delivery worker.

available_events class-attribute

available_events: Sequence[str] | None = None

Event types clients may subscribe to (None: any); patterns must match one.

ninja_devx.contrib.webhooks.network

Outbound URL safety for webhooks: no requests to internal networks (SSRF).

Endpoint URLs come from API clients, so a URL like http://169.254.169.254/ (cloud metadata), http://localhost:8000/admin or http://10.0.0.5/ would make the worker call internal services. Two layers stop that:

• check_url rejects bad schemes, credentials in URLs, and literal private addresses or localhost names when an endpoint is saved (no DNS lookup, so it works offline);
• SafeHTTPTransport checks the address it actually connected to, after DNS resolution, so a public name resolving to a private address (DNS rebinding) is refused too. Proxies from the environment are ignored.

UnsafeURL

Bases: OSError

The URL or the address it resolves to is not allowed.

URLPolicy dataclass

Where webhooks may be sent.

allow_http class-attribute instance-attribute

allow_http: bool = False

Accept http:// URLs (default: HTTPS only).

allow_private_networks class-attribute instance-attribute

allow_private_networks: bool = False

Accept loopback, private, link-local and other non-public addresses (development).

SafeHTTPTransport dataclass

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

policy class-attribute instance-attribute

policy: URLPolicy = URLPolicy()

Where requests may go.

is_public_address

is_public_address(address: str) -> bool

Whether address is a globally routable unicast IP (IPv4-mapped IPv6 unwrapped).

check_url

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

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

Parameters:

Name	Type	Description	Default
url	str	The endpoint URL.	required
policy	URLPolicy | None	What is allowed.	None

ninja_devx.contrib.webhooks.secrets

Encrypting webhook signing secrets at rest.

Signing needs the raw secret, so it cannot be hashed like an API key. With NINJA_DEVX["WEBHOOK_SECRET_KEYS"] set (pip install "ninja-devx[crypto]"), secrets are stored as fernet:<token>; a database dump alone no longer lets anyone forge webhooks. Generate a key with manage.py devx_webhooks generate-key.

Rotation: put the new key first, keep the old ones after it, and run manage.py devx_webhooks encrypt-secrets to re-encrypt with the new key.

encrypt_secret

encrypt_secret(raw: str) -> str

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

Parameters:

Name	Type	Description	Default
raw	str	The signing secret (whsec_...).	required

decrypt_secret

decrypt_secret(stored: str) -> str

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

Parameters:

Name	Type	Description	Default
stored	str	The database value.	required

reencrypt_secret

reencrypt_secret(stored: str) -> str

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

Parameters:

Name	Type	Description	Default
stored	str	The database value.	required

generate_key

generate_key() -> str

A new Fernet key for WEBHOOK_SECRET_KEYS.

ninja_devx.contrib.webhooks.maintenance

Bounded retention and retry operations for workers, APIs and administrators.

retry_deliveries

retry_deliveries(
    deliveries: QuerySet[WebhookDelivery],
) -> int

Reset unclaimed rows; a running worker's ownership is never revoked.

prune_events

prune_events(
    *,
    days: int = 30,
    limit: int = 1000,
    using: str = "default",
) -> int

Delete at most limit old events whose deliveries are all terminal.

Lock deliveries before testing their state, so a concurrent retry cannot be deleted after becoming pending. Events with even one pending delivery are retained.

ninja_devx.contrib.uploads

Presigned uploads with durable ownership and completion tracking.

FakeSigner dataclass

In-memory signer for tests: signer.store(key, size, content_type) simulates the client's upload.

S3Signer dataclass

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

bucket instance-attribute

bucket: str

Bucket name.

prefix class-attribute instance-attribute

prefix: str = 'uploads/'

Key prefix for every upload of this signer.

client class-attribute instance-attribute

client: S3Client | None = None

A boto3 S3 client (default: boto3.client("s3"), created on first use).

UploadController

Bases: Controller

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

signer class-attribute

signer: UploadSigner | None = None

Storage backend (S3Signer, FakeSigner or your own).

policy class-attribute

policy: UploadPolicy = UploadPolicy()

Allowed types, size and expiry.

upload_database class-attribute

upload_database: str = 'default'

Alias containing core UploadRecord rows (run migrations on it).

key_prefix class-attribute

key_prefix: str = ''

Added before <user id>/<uuid>/<filename> (after the signer's own prefix).

owner_prefix

owner_prefix(request: HttpRequest) -> str

The part of the key reserved to the caller (override for tenants).

cleanup_expired

cleanup_expired(*, limit: int = 100) -> int

Delete current objects from expired pending uploads; completed records are retained.

UploadPolicy dataclass

What may be uploaded.

content_types class-attribute instance-attribute

content_types: Sequence[str] = ('*/*',)

Allowed media types; "image/*" allows a family.

max_bytes class-attribute instance-attribute

max_bytes: int = 10 * 1024 * 1024

Largest accepted size.

expires_in class-attribute instance-attribute

expires_in: timedelta = timedelta(minutes=10)

How long the signed form stays valid.

cleanup_grace class-attribute instance-attribute

cleanup_grace: timedelta = timedelta(minutes=15)

Grace after form expiry, allowing in-flight uploads to finish before cleanup.

require_checksum class-attribute instance-attribute

require_checksum: bool = False

Require a base64 SHA-256 digest in the signed request and stored object.

require_version class-attribute instance-attribute

require_version: bool = False

Require storage versioning; consumers must read the returned version_id.

UploadSigner

Bases: Protocol

Signs uploads for a storage backend and inspects what was stored.

safe_filename

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

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

Parameters:

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

ninja_devx.contrib.otel

OpenTelemetry tracing (one span per operation call) and HTTP server metrics.

::

options = ControllerOptions(hooks=[OpenTelemetryHook()])
use_middleware(api, OpenTelemetryMetricsMiddleware())

OpenTelemetryHook dataclass

tracer class-attribute instance-attribute

tracer: Tracer | None = None

Tracer starting one span per call (default: the global tracer).

OpenTelemetryMetricsMiddleware

Bases: Middleware

HTTP server metrics per operation (OpenTelemetry semantic conventions).

http.server.request.duration (histogram, seconds) and http.server.active_requests (up-down counter), with http.request.method, http.route and http.response.status_code. Install on an API or router::

use_middleware(api, OpenTelemetryMetricsMiddleware())

ninja_devx.contrib.tasks

TaskQueue adapters for Celery, Dramatiq, RQ, Taskiq, Temporal and FastStream.

The built-in :class:~ninja_devx.layers.OnCommitTaskQueue defers work through django.tasks. These adapters let services keep depending on the TaskQueue protocol while a framework executes the work.

Framework tasks are callables exposing a deferral method: Celery .delay, Dramatiq .send and Taskiq .kiq are detected automatically. RQ and Temporal need a queue/client, passed to the constructor::

container.singleton(TaskQueue, CeleryTaskQueue())
container.singleton(TaskQueue, RQTaskQueue(redis_queue))
container.singleton(TaskQueue, TemporalTaskQueue(temporal_client))
container.singleton(TaskQueue, FastStreamTaskQueue(broker, queue="tasks"))

DeferredTaskQueue adapts any defer(function, args, kwargs) callable when no framework-specific adapter fits.

DeferredTaskQueue

TaskQueue over any defer(function, args, kwargs) callable.

Use it to adapt an async broker or a custom transport without a framework-specific adapter.

FastStreamTaskQueue

Bases: DeferredTaskQueue

FastStream: publish {"task", "args", "kwargs"} through a broker.

broker.publish is async; the adapter wraps it with async_to_sync so a sync service can enqueue. Pass a FastStream broker (KafkaBroker, NatsBroker, ...)::

from faststream.nats import NatsBroker

container.singleton(TaskQueue, FastStreamTaskQueue(broker, queue="tasks"))

FrameworkTaskQueue

Defer work using a task framework's callable conventions.

Parameters:

Name	Type	Description	Default
queue	object | None	An RQ-style queue exposing enqueue(function, *args, **kwargs).	None
client	object | None	A Temporal-style client exposing start_workflow(workflow, *args).	None
methods	Sequence[str] | None	Attribute names tried on the target callable, in order.	None

CeleryTaskQueue

Bases: FrameworkTaskQueue

Celery: task.delay(*args, **kwargs).

DramatiqTaskQueue

Bases: FrameworkTaskQueue

Dramatiq: actor.send(*args, **kwargs).

TaskiqTaskQueue

Bases: FrameworkTaskQueue

Taskiq: task.kiq(*args, **kwargs).

RQTaskQueue

Bases: FrameworkTaskQueue

RQ: queue.enqueue(function, *args, **kwargs).

TemporalTaskQueue

Bases: FrameworkTaskQueue

Temporal: client.start_workflow(workflow, *args, **kwargs).

ninja_devx.contrib.jobs.runner

Starting, registering and running background jobs.

::

@job("reports.export")
def export_report(ctx: JobContext, report_id: int) -> dict[str, object]:
    ctx.set_progress(50)
    return {"rows": 1000}

@post("/reports/{pk}/export", response={202: dict})
def export(self, request: HttpRequest, pk: int) -> JsonResponse:
    row = start_job(request, "Export report", export_report, report_id=pk)
    return accepted(request, row, prefix="/jobs")

start_job records the call (function name and arguments) on the Job row and enqueues run_job through a TaskQueue, so nothing is pickled: a worker (or devx_jobs retry) resolves the function by the name it was registered or imported with. @job is optional for functions importable by a dotted path; use it for closures, local functions, or a stable name independent of where the function lives.

JobFunction module-attribute

JobFunction = Callable[..., object]

A registered job: (ctx: JobContext, *args, **kwargs) -> object.

JobContext

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

Parameters:

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

set_progress

set_progress(percent: int) -> None

Record how far the job has gotten; call from inside the function.

Parameters:

Name	Type	Description	Default
percent	int	A value from 0 to 100.	required

cancelled

cancelled() -> bool

Whether POST /{id}/cancel was called; long-running functions should check it between steps and stop early.

job

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

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

Parameters:

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

resolve_job

resolve_job(target: str) -> JobFunction

The function registered, or importable, as target.

Parameters:

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

start_job

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.

Parameters:

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

run_job

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.

Never raises: a failure is stored as error and status failed instead of propagating, so a task framework will not retry the call itself.

Parameters:

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

accepted

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

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

Parameters:

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

ninja_devx.contrib.jobs.api

Checking on and cancelling jobs started with start_job.

JobOut

Bases: Schema

id instance-attribute

id: UUID

Primary key, also the value start_job/accepted return in the Location.

name instance-attribute

name: str

Recorded label, shown in the API and admin.

status instance-attribute

status: str

One of Job.Status: queued, running, succeeded, failed, cancelled.

progress instance-attribute

progress: int

Last value reported through JobContext.set_progress (0 to 100).

result instance-attribute

result: dict[str, object]

The function's return value, once succeeded.

error instance-attribute

error: str

The failure message, once failed.

attempts instance-attribute

attempts: int

Times the job has started running, including retries.

created instance-attribute

created: datetime

When start_job created the row.

started instance-attribute

started: datetime | None

When the job most recently started running.

finished instance-attribute

finished: datetime | None

When the job reached a terminal status.

JobsController

Bases: Controller

GET / (the caller's jobs, paginated), GET /{id} and POST /{id}/cancel.

Mount next to the endpoints that call start_job::

api.add_router("/jobs", JobsController.as_router())

GET /{id} and cancelling are open to the job's owner or staff; the list only ever shows the caller's own jobs.

ninja_devx.contrib.jobs.models

Persistence for background jobs started from a request and finished by a worker.

Job

Bases: Model

One run of a function started with start_job and executed by run_job.

ninja_devx.contrib.redis_throttle

Exact, atomic fixed-window throttle counting on Redis.

Needs pip install ninja-devx[redis]. Point a throttle's own storage=, or the project-wide NINJA_DEVX["THROTTLE_STORAGE"], at an instance::

# myapp/throttling.py
from ninja_devx.contrib.redis_throttle import RedisThrottleStorage

redis_throttle_storage = RedisThrottleStorage(url="redis://cache:6379/1")

NINJA_DEVX = {"THROTTLE_STORAGE": "myapp.throttling.redis_throttle_storage"}

INCR and a conditional EXPIRE (NX: only on the window's first hit) run in one MULTI/EXEC transaction, so counting is atomic even under concurrent requests, unlike the default cache-based storage's unprotected add then incr.

RedisThrottleStorage

ThrottleStorage counting fixed windows on Redis with one atomic round trip.

__init__

__init__(
    client: _RedisClient | None = None,
    *,
    url: str = "redis://localhost:6379/0",
) -> None

Parameters:

Name	Type	Description	Default
client	_RedisClient | None	An existing redis.Redis (or redis.cluster.RedisCluster) instance; default: one created from url.	None
url	str	Connection URL used when client is not given.	'redis://localhost:6379/0'

ninja_devx.contrib.nplusone

Runtime N+1 detection: an adapter over django-zeal <https://github.com/taobojlen/django-zeal>_.

Detection stays entirely in zeal; nothing here re-implements query counting. Install the middleware so every request runs inside zeal's tracking context (zeal.setup()/ zeal.teardown(), the primitives zeal.zeal_context() wraps), and when zeal raises NPlusOneError the middleware rewrites its message to name the controller and point at the fix::

install(api, [NPlusOnePlugin()])

or, without the plugin system::

use_middleware(api, NPlusOneMiddleware())

Requires "zeal" in INSTALLED_APPS and the zeal extra (pip install ninja-devx[zeal], django-zeal>=2); tune detection with zeal's own settings (ZEAL_RAISE, ZEAL_NPLUSONE_THRESHOLD, ZEAL_ALLOWLIST, ...). Everything here is a no-op when zeal is not installed, so it is safe to add unconditionally.

NPlusOneMiddleware

Bases: Middleware

Runs the request inside zeal's tracking context and explains its errors.

A no-op when zeal is not installed.

NPlusOnePlugin

Bases: APIPlugin

Adds :class:NPlusOneMiddleware to every operation of the API.

zeal_installed

zeal_installed() -> bool

Whether django-zeal is importable.

explain_n_plus_one

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

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

zeal's own message is "N+1 detected on <app>.<Model>.<field> at <file>:<line> in <function>"; this appends the controller (when known) and the ninja_devx.crud.optimization names that fix it. Messages zeal did not produce (no model.field) are returned unchanged.

Parameters:

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

zeal_strict

zeal_strict() -> Generator[None]

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

Backs the strict_queries pytest fixture (ninja_devx.testing). Call :func:zeal_installed first: this assumes zeal is installed and does not skip.

Tooling

ninja_devx.testing.clients

Test helpers: exercise controllers through Django Ninja's test clients.

AuthenticatedClient

Bases: TestClient

A TestClient that sends every request as user unless one is given.

AuthenticatedAsyncClient

Bases: TestAsyncClient

A TestAsyncClient that sends every request as user unless one is given.

HopCounter dataclass

Thread hops (sync_to_async calls) seen inside assert_max_hops.

client_for

client_for(
    controller: type[Controller],
    *,
    user: object | None = None,
    container: ContainerLike | None = None,
    scope: Scope | None = None,
    **options: Unpack[ControllerOptions],
) -> AuthenticatedClient

A test client for controller.as_router(...); paths are relative to the router.

user becomes request.user for every request (override per request with user=).

Parameters:

Name	Type	Description	Default
controller	type[Controller]	The controller class.	required
user	object | None	Sent as request.user with every request (override per request with user=).	None
container	ContainerLike | None	Container for as_router().	None
scope	Scope | None	Scope for as_router().	None
options	Unpack[ControllerOptions]	ControllerOptions for as_router().	{}

async_client_for

async_client_for(
    controller: type[Controller],
    *,
    user: object | None = None,
    container: ContainerLike | None = None,
    scope: Scope | None = None,
    **options: Unpack[ControllerOptions],
) -> AuthenticatedAsyncClient

Parameters:

Name	Type	Description	Default
controller	type[Controller]	The controller class.	required
user	object | None	Sent as request.user with every request.	None
container	ContainerLike | None	Container for as_router().	None
scope	Scope | None	Scope for as_router().	None
options	Unpack[ControllerOptions]	ControllerOptions for as_router().	{}

assert_max_queries

assert_max_queries(
    limit: int, *, using: str = DEFAULT_DB_ALIAS
) -> Generator[CaptureQueriesContext]

Fail when the block runs more than limit queries (catches N+1 regressions).

Parameters:

Name	Type	Description	Default
limit	int	Maximum number of queries in the block.	required
using	str	Database alias to watch.	DEFAULT_DB_ALIAS

assert_max_hops

assert_max_hops(limit: int) -> Generator[HopCounter]

Fail when the block switches from the event loop to a thread more than limit times.

Each hop costs tens of microseconds and serializes on the request's thread::

with assert_max_hops(1):
    await client.post("/", json=payload)   # async CRUD create: one hop

Parameters:

Name	Type	Description	Default
limit	int	Maximum sync_to_async thread hops in the block.	required

capture_commits

capture_commits(
    *, using: str = DEFAULT_DB_ALIAS, execute: bool = True
) -> Generator[list[Callable[[], object]]]

Collect transaction.on_commit callbacks (after_commit, OnCommitTaskQueue).

In a test transaction nothing ever commits; with execute=True the callbacks run when the block exits, as they would after a real commit::

with capture_commits() as callbacks:
    client.post("/orders", json=payload)
assert len(callbacks) == 1          # the confirmation email was scheduled

Parameters:

Name	Type	Description	Default
using	str	Database alias.	DEFAULT_DB_ALIAS
execute	bool	Run the callbacks when the block exits.	True

ninja_devx.testing.factories

Synthetic request payloads and records for tests.

sample(Schema) builds a valid, JSON-serialisable payload from a pydantic schema: declared defaults and examples win, otherwise a value is derived from the field type. samples(Schema, n) varies strings and numbers so rows are distinguishable::

from ninja_devx.testing import sample, samples

payload = sample(ArticleIn)                 # {"title": "title", ...}
rows = samples(ArticleIn, 3)                # three distinct payloads

A required field whose type has no derivation rule raises TypeError; give it an example or a default instead.

sample

sample(schema: type[SchemaT]) -> dict[str, object]

A payload for schema built from defaults, examples and field types.

Parameters:

Name	Type	Description	Default
schema	type[SchemaT]	The pydantic schema (usually the request body).	required

samples

samples(
    schema: type[SchemaT], count: int
) -> list[dict[str, object]]

count payloads for schema with distinguishable scalar values.

Parameters:

Name	Type	Description	Default
schema	type[SchemaT]	The pydantic schema (usually the request body).	required
count	int	How many payloads to build.	required

ninja_devx.testing.plugin

pytest plugin (auto-loaded): controller clients and OpenAPI snapshots.

Fixtures:

• ninja_client(ControllerOrRouter, user=None, **as_router_kwargs) -> test client
• ninja_async_client(...) -> async test client
• ninja_contract(api, include=None) -> a schemathesis schema calling the app in-process (pip install ninja-devx[contract])
• captured_commits(execute=True) -> ninja_devx.testing.clients.capture_commits
• openapi_snapshot(api, name="openapi") compares api's schema with __snapshots__/<test module>/<name>.json; run pytest --update-snapshots to accept changes.
• strict_queries runs the test inside django-zeal, raising on any N+1; skips with a clear reason when zeal (the ninja-devx[zeal] extra) is not installed.

Async tests that use the database without django_db(transaction=True) get an AsyncDatabaseTestWarning: their ORM calls run on another thread and connection, outside the test transaction. Disable with ninja_devx_warn_async_db = false (ini).

Query-count fixtures come from pytest-django (django_assert_max_num_queries). Django and Ninja are imported lazily: plugins load before settings are configured.

ninja_client

ninja_client() -> Callable[..., AuthenticatedClient]

ninja_client(ControllerOrRouter, user=None, container=None, scope=None, **options).

ninja_async_client

ninja_async_client() -> Callable[
    ..., AuthenticatedAsyncClient
]

ninja_async_client(ControllerOrRouter, user=None, **options): an async test client.

ninja_contract

ninja_contract() -> Callable[..., object]

ninja_contract(api): a schemathesis schema for schemathesis.pytest.from_fixture.

captured_commits

captured_commits() -> Callable[
    ..., AbstractContextManager[list[Callable[[], object]]]
]

with captured_commits() as callbacks: ... runs on-commit work at the end.

openapi_snapshot

openapi_snapshot(
    request: FixtureRequest,
) -> Callable[..., None]

openapi_snapshot(api, name="openapi") compares the schema with a stored snapshot.

strict_queries

strict_queries() -> Generator[None]

Raise zeal.NPlusOneError for any N+1 in this test, regardless of ZEAL_RAISE.

Skips with a clear reason when django-zeal is not installed (pip install ninja-devx[zeal]).

ninja_devx.configuration.checks

Django system checks for controllers: manage.py check reports them in CI.

Checked controllers are those behind the APIs listed in NINJA_DEVX["CHECK_APIS"] (dotted paths to NinjaAPI instances), or, when it is empty, every router built by as_router() once the URLconf is loaded.

Messages come from Controller.checks() (overridable) and from plugins that define checks(controller). Startup errors that make a controller unusable are still raised by as_router() itself.

======================== =========================================================== ninja_devx.E001 a CHECK_APIS entry cannot be imported or is not an API ninja_devx.E002 search_fields/filter_fields/ordering_fields name a missing model field ninja_devx.W003 an output schema field is neither a model field, a model attribute nor resolved by the schema ninja_devx.E004 service_class needs constructor arguments but the controller is mounted without a container ninja_devx.W005 an output field VisibleTo can hide is required instead of optional ninja_devx.W006 a related/@requires_related hint is not a relation path of the model ninja_devx.W007 expand_rules[...].limit targets a relation that is not a plain reverse foreign key ninja_devx.E007 aggregate_fields names a missing model field ======================== ===========================================================

CHECKS module-attribute

CHECKS: Final[Mapping[str, tuple[str, str]]] = (
    MappingProxyType(
        {
            "ninja_devx.E001": (
                "error",
                "A ``CHECK_APIS`` entry cannot be imported or is not a ``NinjaAPI``.",
            ),
            "ninja_devx.E002": (
                "error",
                "``search_fields``, ``filter_fields``, ``ordering_fields`` or ``default_ordering`` names a missing model field.",
            ),
            "ninja_devx.W003": (
                "warning",
                "An output schema field is not a model field or attribute, and the schema does not resolve it.",
            ),
            "ninja_devx.E004": (
                "error",
                "``service_class`` needs constructor arguments, but the controller has no container.",
            ),
            "ninja_devx.W005": (
                "warning",
                "An output field that ``VisibleTo`` can hide is required instead of optional.",
            ),
            "ninja_devx.W006": (
                "warning",
                "A ``related`` hint or ``@requires_related`` lookup is not a relation of the model.",
            ),
            "ninja_devx.W007": (
                "warning",
                "``expand_rules[...].limit`` targets a relation that is not a plain reverse foreign key, so it is prefetched without a cap.",
            ),
            "ninja_devx.E007": (
                "error",
                "``aggregate_fields`` names a field that does not exist on the model.",
            ),
        }
    )
)

Every system check id with its level and meaning (also rendered in the docs).

load_urlconf

load_urlconf() -> None

Import the URLconf so every API and controller module is loaded.

ninja_devx.tooling.inspect

Resolve a mounted controller into its effective policy, for humans and tooling.

devx_inspect prints the tree; inspect_target returns the same data as dataclasses so it can be rendered as JSON or asserted in tests. Everything is read from the live controller and its settings, never from a static declaration.

OperationInspection dataclass

One registered operation of a controller.

ControllerInspection dataclass

The resolved configuration and operations of one mounted controller.

resolve

resolve(
    target: str | None,
) -> list[tuple[str, BuiltRouter]]

Mounted controllers matching target (a controller path or a route prefix).

Raises:

Type	Description
LookupError	target is a route prefix but no CHECK_APIS are configured, so prefixes are unknown.

inspect_target

inspect_target(
    target: str | None = None,
) -> list[ControllerInspection]

Inspect every mounted controller matching target (all of the APIs without one).

as_dict

as_dict(
    results: Iterable[ControllerInspection],
) -> list[dict[str, object]]

The inspections as plain dictionaries, ready for json.dumps.

render

render(inspection: ControllerInspection) -> str

A box-drawing tree of one controller inspection.

ninja_devx.tooling.openapi_diff

Semantic OpenAPI diff for CI: what changed, and what breaks clients.

devx_openapi --against baseline.json fails on breaking changes. Additive changes (new paths, new optional fields, new enum values) are reported but do not fail.

Change dataclass

One difference between two documents: severity is BREAKING or ADDITIVE.

diff

diff(baseline: Schema, current: Schema) -> list[Change]

Compare two OpenAPI documents, reporting breaking and additive changes.

has_breaking

has_breaking(changes: Iterable[Change]) -> bool

Whether any change breaks existing clients.

ninja_devx.tooling.drift

Detect drift between models and the schemas of the controllers exposing them.

manage.py devx_scaffold --check runs this for every mounted ModelController (or the models given) and fails when a schema no longer matches its model::

NoteController  NoteOut.priority   error    model IntegerField is int, schema says str
NoteController  NoteIn             error    required field 'title' is not accepted
NoteController  NoteOut            info     model field 'archived' is not exposed

Only clear incompatibilities are errors: a value the model can hold that the schema would reject (type, None, choices), a required field create cannot fill, or a schema field the model does not have.

Drift dataclass

controller instance-attribute

controller: str

Controller name.

location instance-attribute

location: str

Schema or schema field.

severity instance-attribute

severity: Severity

error, warning or info.

message instance-attribute

message: str

What is wrong.

schema_drift

schema_drift(
    controller: type[ModelController[Model]],
) -> list[Drift]

Mismatches between controller's model and its output/input schemas.

ninja_devx.tooling.scaffold

Generate explicit, typed schemas, a controller and tests from a Django model.

Used by manage.py devx_scaffold. The output is plain Python that type checkers understand, so generated schemas can be used as generic arguments.

render_resource

render_resource(
    model: type[Model],
    options: ScaffoldOptions | None = None,
) -> str

Schemas (<Model>In/<Model>Out) and a controller for model.

ninja_devx.tooling.startproject

Post-render edits for manage.py devx_startproject that Django's template engine cannot express as a plain variable substitution: wiring a first app into the generated settings and API module, and dropping the Docker files for --no-docker.

Plain string edits over the rendered project, not a second templating pass::

wire_app(directory, "blog")   # after ``call_command("devx_startapp", "blog", ...)``
drop_docker(directory)        # deletes compose.yaml and its README section

camel_case

camel_case(name: str) -> str

blog_posts -> BlogPosts, matching devx_startapp's controller names.

wire_app

wire_app(directory: Path, app_name: str) -> None

Add app_name to INSTALLED_APPS and mount its controller in config/api.py.

Call after devx_startapp has written app_name under directory.

Parameters:

Name	Type	Description	Default
directory	Path	The generated project's root.	required
app_name	str	The app just scaffolded by devx_startapp.	required

drop_docker

drop_docker(directory: Path) -> None

Remove compose.yaml and the matching section of README.md for --no-docker.

Parameters:

Name	Type	Description	Default
directory	Path	The generated project's root.	required

target_directory

target_directory(directory: Path) -> Generator[None]

Create directory for Django's startproject/startapp, which only create a missing destination themselves from Django 6.0.

The directory is removed again if the command fails before writing into it.

ninja_devx.tooling.doctor

Findings beyond the system checks, across every mounted controller.

devx_doctor inspects the same mounted controllers as manage.py check (:mod:ninja_devx.configuration.checks), through :func:ninja_devx.tooling.inspect.resolve, but looks for things that are valid yet risky rather than broken: a filter, search or ordering field with no database index, an owner_field/tenant_field with nothing actually enforcing it, an output field the N+1 planner cannot resolve statically, a list endpoint with no pagination, a controller with no permission at all, and a soft-deleted model whose unique fields are not scoped to active rows.

Each check is one small function added to :data:CHECKS, so adding a new one is one function::

@_register
def _check_something(target: _Target) -> Iterator[Finding]:
    if ...:
        yield Finding("warn", target.inspection.controller, "message", "fix hint")

:func:run_doctor runs every registered check over every controller resolve(target) finds.

Finding dataclass

One thing devx_doctor noticed about a mounted controller.

severity instance-attribute

severity: Severity

"info" or "warn".

controller instance-attribute

controller: str

Controller name, from :attr:ControllerInspection.controller.

message instance-attribute

message: str

What was noticed.

hint instance-attribute

hint: str

How to address it.

run_doctor

run_doctor(target: str | None = None) -> list[Finding]

Run every registered check over every controller resolve(target) finds.

Parameters:

Name	Type	Description	Default
target	str | None	A controller path or a mounted route prefix; every controller of the configured APIs when None (see :func:ninja_devx.tooling.inspect.resolve).	None

ninja_devx.tooling.unasync

Generate the sync twin of async code, and check that it has not drifted.

Write the async implementation once; the sync one is derived token by token (async def → def, await x → x, async for/with → for/with, aget → get, AsyncIterator → Iterator...)::

python -m ninja_devx.tooling.unasync src/orders/aio.py:src/orders/sync.py
python -m ninja_devx.tooling.unasync src/orders/aio.py:src/orders/sync.py --check   # CI
python -m ninja_devx.tooling.unasync a.py:b.py --replace AsyncOrderRepository=OrderRepository

Names are replaced as whole tokens (strings and comments are left alone, except the generated-file header). Django's async ORM methods, this package's a-prefixed helpers and the standard async protocol names are mapped by default.

unasync_source

unasync_source(
    code: str,
    replacements: Mapping[str, str] = DEFAULT_REPLACEMENTS,
    *,
    header: str = "",
) -> str

The sync version of code.

ninja_devx.testing.contracts

Contract tests generated from the OpenAPI schema, with schemathesis.

Install the extra (pip install ninja-devx[contract]), then::

import schemathesis

@pytest.fixture
def api_schema(ninja_contract, db):
    return ninja_contract(api)

schema = schemathesis.pytest.from_fixture("api_schema")     # .include(path_regex=...) to filter

@schema.parametrize()
def test_api_contract(case):
    case.call_and_validate(headers={"Authorization": "Bearer test-token"})

Every operation gets property-based requests (valid and invalid input). Responses are checked against the documented status codes, content types and schemas, so an undocumented 500 or a body that doesn't match response= fails the test. Requests go through Django's WSGI application in-process, inside the test database transaction.

contract_schema

contract_schema(
    api: NinjaAPI, *, path_prefix: str | None = None
) -> OpenApiSchema

A schemathesis schema for api, calling the Django WSGI app in-process.

path_prefix defaults to where the API is mounted in the URLconf. Filter operations on the lazy schema: schemathesis.pytest.from_fixture("api_schema").include(...).

Parameters:

Name	Type	Description	Default
api	NinjaAPI	The NinjaAPI to test.	required
path_prefix	str | None	API root path (default: where it is mounted in the URLconf).	None

ninja_devx.codegen

Typed API clients generated from a Django Ninja OpenAPI schema.

Operation dataclass

response instance-attribute

response: Schema | None

Union of all documented 2xx JSON, text and binary representations.

has_body instance-attribute

has_body: bool

Whether any documented success response has a body.

errors class-attribute instance-attribute

errors: tuple[ResponseVariant, ...] = ()

Documented non-2xx responses that carry a body.

streaming class-attribute instance-attribute

streaming: bool = False

A 2xx text/event-stream response, generated as an event iterator.

load_api

load_api(path: str) -> NinjaAPI

Import a NinjaAPI from a dotted path, e.g. config.urls.api.

ninja_devx.contrib.dishka

Use a dishka <https://dishka.readthedocs.io>_ container with controllers.

Every request opens a dishka scope (Scope.REQUEST by default) in which HttpRequest is available as context; declare it with from_context(provides=HttpRequest, scope=Scope.REQUEST)::

provider = AppProvider()
provide_controllers(provider, [UserController, OrderController])

resolver = DishkaResolver(
    make_container(provider),
    async_container=make_async_container(provider),   # only for async operations
)
api.add_router("/users", UserController.as_router(container=resolver))

One resolver serves sync and async operations. as_router() fails at startup when the controller (or an Inject[T]) is not provided, instead of on the first request.

DishkaResolver dataclass

Sync operations use container; async ones async_container when given.

container instance-attribute

container: Container

The sync dishka container.

scope class-attribute instance-attribute

scope: BaseScope = Scope.REQUEST

dishka scope opened per request.

async_container class-attribute instance-attribute

async_container: AsyncContainer | None = None

Async container for async operations (optional).

resolve

resolve(key: type[T]) -> T

Resolve from the root container (singleton controllers).

AsyncDishkaResolver dataclass

An async-only resolver: every operation using it must be async.

container instance-attribute

container: AsyncContainer

The async dishka container.

scope class-attribute instance-attribute

scope: BaseScope = Scope.REQUEST

dishka scope opened per request.

provide_controllers

provide_controllers(
    provider: Provider,
    controllers: Iterable[type[object]],
    *,
    scope: BaseScope = Scope.REQUEST,
) -> None

Register controller classes with provider (their __init__ is autowired).

Parameters:

Name	Type	Description	Default
provider	Provider	The dishka provider to register on.	required
controllers	Iterable[type[object]]	Controller classes (their __init__ is autowired).	required
scope	BaseScope	dishka scope of the controllers.	REQUEST

ninja_devx.contrib.svcs

Use an svcs <https://svcs.hynek.me>_ registry with controllers.

Every request gets its own svcs.Container (closed, with cleanups, when the request ends) in which HttpRequest is registered as a local value::

registry = svcs.Registry()
registry.register_factory(Database, connect)
api.add_router("/users", UserController.as_router(container=SvcsResolver(registry)))

svcs containers are per request, so controllers must use Scope.REQUEST.

SvcsResolver dataclass

registry instance-attribute

registry: Registry

The svcs registry.