APIViewSet¶
APIViewSet auto-generates async CRUD endpoints and optional Many-to-Many (M2M) endpoints for a Django Model or a ModelSerializer. It supports dynamic schema generation, per-verb authentication, pagination, list & relation filtering with runtime-built Pydantic schemas, and custom view injection.
Generated CRUD Endpoints¶
| Method | Path | Summary | Response |
|---|---|---|---|
| POST | /{base}/ |
Create Model | 201 schema_out |
| GET | /{base}/ |
List Models | 200 List[schema_out] (paginated) |
| GET | /{base}/{pk} |
Retrieve Model | 200 schema_detail (or schema_out if none) |
| PATCH | /{base}/{pk}/ |
Update Model | 200 schema_out |
| DELETE | /{base}/{pk}/ |
Delete Model | 204 No Content |
Notes:
- Retrieve path typically includes a trailing slash by default (see settings below); update/delete include a trailing slash.
{base}auto-resolves from model verbose name plural (lowercase) unlessapi_route_pathis provided.- Error responses may use a unified generic schema for codes: 400, 401, 404.
Settings: trailing slash behavior¶
- NINJA_AIO_APPEND_SLASH (default: True)
- When True (default, for backward compatibility), retrieve and POST paths includes a trailing slash into CRUD:
/{base}/{pk}/. - When False, retrieve and post paths is generated without a trailing slash:
/{base}/{pk}.
Request Lifecycle¶
graph TD
A[Incoming Request] --> B{Auth Check}
B -->|Fail| X[401 Unauthorized]
B -->|Pass| C{Route Match}
C -->|List| D[Apply Filters]
C -->|Create/Update| E[Parse & Validate Schema]
C -->|Delete| F[Lookup Object]
D --> G[Apply Ordering]
G --> H[Paginate QuerySet]
H --> I[Serialize Response]
E --> J[Run Hooks & Save]
J --> I
F --> K[Delete & Return 204]
I --> L[JSON Response]
style A fill:#7c4dff,stroke:#7c4dff,color:#fff
style B fill:#ff6d00,stroke:#ff6d00,color:#fff
style X fill:#d50000,stroke:#d50000,color:#fff
style L fill:#00c853,stroke:#00c853,color:#fff
style K fill:#00c853,stroke:#00c853,color:#fff
Recommended: @action Decorator¶
The @action decorator is the recommended way to add custom endpoints to your ViewSet. It provides automatic URL generation, auth inheritance, detail/list distinction, and full OpenAPI metadata support.
Basic Usage¶
from ninja import Schema, Status
from ninja_aio.decorators import action
class CountSchema(Schema):
count: int
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
@action(detail=True, methods=["post"], url_path="activate")
async def activate(self, request, pk):
obj = await self.model_util.get_object(request, pk)
obj.is_active = True
await obj.asave()
return Status(200, {"message": "activated"})
@action(detail=False, methods=["get"], url_path="count", response=CountSchema)
async def count(self, request):
total = await self.model.objects.acount()
return {"count": total}
Parameters¶
| Parameter | Type | Default | Description |
|---|---|---|---|
detail |
bool |
— (required) | True = instance action (/{pk}/path), False = collection action |
methods |
list[str] |
["get"] |
HTTP methods to register |
url_path |
str \| None |
None |
Custom URL segment. Defaults to method name with _ → - |
url_name |
str \| None |
None |
Django URL name for reverse resolution |
auth |
Any |
NOT_SET |
Auth override. NOT_SET inherits from viewset per-verb auth |
response |
Any |
NOT_SET |
Response schema. NOT_SET lets Django Ninja infer it |
summary |
str \| None |
None |
OpenAPI summary (auto-generated if None) |
description |
str \| None |
None |
OpenAPI description |
tags |
list[str] \| None |
None |
OpenAPI tags (inherits from viewset if None) |
deprecated |
bool \| None |
None |
Mark as deprecated in OpenAPI |
decorators |
list[Callable] \| None |
None |
Additional decorators to apply |
throttle |
BaseThrottle \| list[BaseThrottle] |
NOT_SET |
Throttle configuration |
include_in_schema |
bool |
True |
Whether to include in OpenAPI schema |
openapi_extra |
dict \| None |
None |
Additional OpenAPI metadata |
Detail vs List Actions¶
Detail actions (detail=True) operate on a single instance. The pk parameter is automatically added to the URL and renamed to match the model's primary key field name:
@action(detail=True, methods=["post"], url_path="publish")
async def publish(self, request, pk):
# URL: /article/{id}/publish
obj = await self.model_util.get_object(request, pk)
...
List actions (detail=False) operate on the collection:
@action(detail=False, methods=["get"], url_path="stats")
async def stats(self, request):
# URL: /article/stats
...
Multiple HTTP Methods¶
Register the same action for multiple HTTP methods:
@action(detail=True, methods=["get", "post"], url_path="toggle")
async def toggle(self, request, pk):
# Registers both GET and POST on /article/{id}/toggle
...
Auth Inheritance¶
Actions inherit authentication from the viewset by default:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
auth = [JWTAuth()]
# Inherits JWTAuth from viewset
@action(detail=False, methods=["get"], url_path="protected")
async def protected(self, request):
...
# Override: make public
@action(detail=False, methods=["get"], url_path="public", auth=None)
async def public(self, request):
...
Custom Decorators¶
Apply additional decorators to action handlers:
from ninja_aio.decorators import aatomic
@action(detail=False, methods=["post"], url_path="batch-op", decorators=[aatomic])
async def batch_operation(self, request):
# Wrapped in atomic transaction
...
URL Path Auto-Generation¶
When url_path is not provided, the method name is used with underscores replaced by hyphens:
@action(detail=False, methods=["get"])
async def my_custom_endpoint(self, request):
# URL: /article/my-custom-endpoint
...
Actions and disable¶
Actions are not affected by disable = ["all"]. Custom actions are always registered even when all CRUD endpoints are disabled:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
disable = ["all"] # No CRUD endpoints
@action(detail=False, methods=["get"], url_path="health")
async def health(self, request):
return {"status": "ok"} # Still registered
Alternative: @api_get / @api_post decorators¶
Class method decorators for adding custom endpoints. These provide direct control over the path and HTTP method, but without the automatic features of @action (no auto {pk}, no auth inheritance, no auto URL path).
Available decorators (from ninja_aio.decorators):
@api_get(path, ...)@api_post(path, ...)@api_put(path, ...)@api_patch(path, ...)@api_delete(path, ...)@api_options(path, ...)@api_head(path, ...)
Example:
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from ninja_aio.decorators import api_get, api_post
from .models import Article
api = NinjaAIO(title="Blog API")
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
@api_get("/stats/")
async def stats(self, request):
total = await self.model.objects.acount()
return {"total": total}
@api_post("/{pk}/publish/")
async def publish(self, request, pk: int):
obj = await self.model.objects.aget(pk=pk)
obj.is_published = True
await obj.asave()
return {"message": "published"}
Notes:
- Decorators support per-endpoint
auth,response,tags,summary,description, and more. - Sync methods are executed via
sync_to_asyncautomatically. - Signatures and type hints are preserved for OpenAPI (excluding
self).
Legacy: views() method (still supported)¶
The previous pattern of injecting endpoints inside views() is still supported, but the @action and @api_* approaches above are now recommended.
class ArticleViewSet(APIViewSet):
model = Article
api = api
def views(self):
@self.router.get("/stats/")
async def stats(request):
total = await self.model.objects.acount()
return {"total": total}
@self.router.post("/{pk}/publish/")
async def publish(request, pk: int):
obj = await self.model.objects.aget(pk=pk)
obj.is_published = True
await obj.asave()
return {"message": "published"}
Core Attributes¶
| Attribute | Type | Default | Description |
|---|---|---|---|
model |
ModelSerializer \| Model |
— | Target model (required) |
api |
NinjaAPI |
— | API instance (required) |
serializer_class |
Serializer \| None |
None |
Serializer class for plain models (alternative to ModelSerializer) |
schema_in |
Schema \| None |
None (auto) |
Create input schema override |
schema_out |
Schema \| None |
None (auto) |
List/output schema override |
schema_detail |
Schema \| None |
None (auto) |
Retrieve/detail schema override (falls back to schema_out) |
schema_update |
Schema \| None |
None (auto) |
Update input schema override |
pagination_class |
type[AsyncPaginationBase] |
PageNumberPagination |
Pagination strategy |
query_params |
dict[str, tuple[type, ...]] |
{} |
List endpoint filters definition |
disable |
list[type[VIEW_TYPES]] |
[] |
Disable views (create,list,retrieve,update,delete,bulk_create,bulk_update,bulk_delete,all) |
api_route_path |
str |
"" |
Base route segment |
list_docs |
str |
"List all objects." |
List endpoint description |
create_docs |
str |
"Create a new object." |
Create endpoint description |
retrieve_docs |
str |
"Retrieve a specific object by its primary key." |
Retrieve endpoint description |
update_docs |
str |
"Update an object by its primary key." |
Update endpoint description |
delete_docs |
str |
"Delete an object by its primary key." |
Delete endpoint description |
m2m_relations |
list[M2MRelationSchema] |
[] |
M2M relation configs |
m2m_auth |
list \| None |
NOT_SET |
Default auth for all M2M endpoints (overridden per relation if set) |
extra_decorators |
DecoratorsSchema |
DecoratorsSchema() |
Custom decorators for CRUD and bulk operations |
model_verbose_name |
str |
"" |
Override model verbose name for display |
model_verbose_name_plural |
str |
"" |
Override model verbose name plural for display |
bulk_operations |
list[Literal["create", "update", "delete"]] |
[] |
Bulk operations to enable (opt-in) |
bulk_response_fields |
list[str] \| str \| None |
None |
Field(s) returned in bulk success details (None = PK) |
bulk_create_docs |
str |
"Create multiple objects in a single request." |
Bulk create endpoint description |
bulk_update_docs |
str |
"Update multiple objects in a single request." |
Bulk update endpoint description |
bulk_delete_docs |
str |
"Delete multiple objects in a single request." |
Bulk delete endpoint description |
ordering_fields |
list[str] |
[] |
Fields allowed for ordering (empty = disabled) |
default_ordering |
str \| list[str] |
[] |
Default ordering when no ?ordering param is provided |
require_update_fields |
bool |
False |
Reject PATCH with empty payload (SerializeError) |
Verbose Name Resolution¶
Verbose names are resolved using a three-tier priority:
| Priority | Source | Example |
|---|---|---|
| 1 | ViewSet class attribute (model_verbose_name) |
Set on the ViewSet directly |
| 2 | NinjaAIOMeta inner class on the model |
class NinjaAIOMeta: verbose_name = "Blog Article" |
| 3 | Django Meta (model._meta.verbose_name) |
Auto-derived from model class name |
Using NinjaAIOMeta on the model (recommended for reuse across ViewSets):
class BlogPost(models.Model):
title = models.CharField(max_length=255)
class NinjaAIOMeta:
not_found_name = "article" # custom 404 error key
verbose_name = "Blog Article" # used for endpoint summaries
verbose_name_plural = "Blog Articles" # used for route paths & display
All attributes are optional. NinjaAIOMeta values are used as-is (no .capitalize()), while Django Meta values are auto-capitalized.
Using ViewSet attributes (per-ViewSet override):
@api.viewset(BlogPost)
class BlogPostAPI(APIViewSet):
model_verbose_name = "Article"
model_verbose_name_plural = "Articles"
ViewSet attributes take highest priority, so they can override NinjaAIOMeta for a specific ViewSet.
Authentication Attributes¶
| Attribute | Type | Default | Description |
|---|---|---|---|
auth |
list \| None |
NOT_SET |
Global fallback auth |
get_auth |
list \| None |
NOT_SET |
Auth for list + retrieve |
post_auth |
list \| None |
NOT_SET |
Auth for create |
patch_auth |
list \| None |
NOT_SET |
Auth for update |
delete_auth |
list \| None |
NOT_SET |
Auth for delete |
Resolution rules:
- Per-verb auth overrides
authwhen notNOT_SET. Nonemakes the endpoint public (no authentication).- M2M endpoints use relation-level auth (
m2m_data.auth) or fall back tom2m_auth.
Transaction Management¶
Create, update, and delete operations are automatically wrapped in atomic transactions using the @aatomic decorator. This ensures that database operations are rolled back on exceptions:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
pass # create/update/delete automatically transactional
The transaction behavior is applied by default. Custom decorators can be added via extra_decorators attribute.
Automatic Schema Generation¶
If model is a subclass of ModelSerializerMeta:
schema_outis generated fromReadSerializerschema_detailis generated fromDetailSerializer(optional, falls back toschema_out)schema_infromCreateSerializerschema_updatefromUpdateSerializer
For plain Django models, you can provide a serializer_class (Serializer) instead:
from ninja_aio.models import serializers
class ArticleSerializer(serializers.Serializer):
class Meta:
model = models.Article
schema_in = serializers.SchemaModelConfig(
fields=["title", "content", "author"]
)
schema_out = serializers.SchemaModelConfig(
fields=["id", "title", "content", "author"]
)
@api.viewset(model=models.Article)
class ArticleViewSet(APIViewSet):
serializer_class = ArticleSerializer
Otherwise provide schemas manually via schema_in, schema_out, schema_detail, and schema_update attributes.
Detail Schema for Retrieve Endpoint¶
Use schema_detail (or DetailSerializer on ModelSerializer) when you want the retrieve endpoint to return more fields than the list endpoint. This is useful for:
- Performance optimization: List endpoints return minimal fields, retrieve endpoints include expensive relations
- API design: Clients get a summary in lists and full details on individual requests
from ninja_aio.models import ModelSerializer
from django.db import models
class Article(ModelSerializer):
title = models.CharField(max_length=200)
summary = models.TextField()
content = models.TextField()
author = models.ForeignKey(User, on_delete=models.CASCADE)
tags = models.ManyToManyField(Tag)
class ReadSerializer:
# List view: minimal fields
fields = ["id", "title", "summary"]
class DetailSerializer:
# Detail view: all fields
fields = ["id", "title", "summary", "content", "author", "tags"]
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
pass # Schemas auto-generated from model
Endpoints behavior:
- GET /articles/ returns [{"id": 1, "title": "...", "summary": "..."}, ...]
- GET /articles/1 returns {"id": 1, "title": "...", "summary": "...", "content": "...", "author": {...}, "tags": [...]}
Partial Update Validation¶
By default, PATCH endpoints accept empty payloads silently (no fields updated, but save() is still called). Enable require_update_fields to reject empty updates:
When enabled, a PATCH request with no update fields raises a SerializeError with message "No fields provided for update." (HTTP 400).
This applies to both single update and bulk update operations. In bulk updates, empty items are collected as errors without affecting other items (partial success semantics).
Bulk Operations¶
Bulk operations allow creating, updating, or deleting multiple objects in a single request. They are opt-in — no bulk endpoints are registered unless explicitly enabled via bulk_operations.
Enabling Bulk Operations¶
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
bulk_operations = ["create", "update", "delete"]
Generated Bulk Endpoints¶
| Method | Path | Summary | Response |
|---|---|---|---|
| POST | /{base}/bulk/ |
Bulk Create | 200 BulkResultSchema |
| PATCH | /{base}/bulk/ |
Bulk Update | 200 BulkResultSchema |
| DELETE | /{base}/bulk/ |
Bulk Delete | 200 BulkResultSchema |
Response Format¶
All bulk endpoints return a BulkResultSchema with partial success semantics — each item is processed independently, and failures don't affect other items:
{
"success": {
"count": 2,
"details": [1, 3]
},
"errors": {
"count": 1,
"details": [{"error": "Not found."}]
}
}
success.details— list of primary keys of successfully processed objects (default). Customizable viabulk_response_fields.errors.details— list of error detail dicts for each failed item.
Custom Response Fields¶
By default, success.details contains primary keys. Use bulk_response_fields to return different field(s):
Single field — returns a flat list of values:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
bulk_operations = ["create", "update", "delete"]
bulk_response_fields = "title"
Multiple fields — returns a list of dicts:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
bulk_operations = ["create", "update", "delete"]
bulk_response_fields = ["id", "title"]
{
"success": {
"count": 2,
"details": [
{"id": 1, "title": "Article 1"},
{"id": 2, "title": "Article 2"}
]
}
}
Note
For bulk delete, the requested fields are fetched before deletion so they can be included in the response.
Request Formats¶
Bulk Create — request body is List[schema_in]:
Bulk Update — request body is a list of objects with the PK field (required) plus update fields. A dynamic schema is generated combining the PK field with schema_update:
Bulk Delete — request body contains a list of PK values:
Selective Enablement¶
Enable only specific operations:
bulk_operations = ["create"] # Only bulk create
bulk_operations = ["create", "delete"] # Create and delete, no update
Disabling Individual Bulk Views¶
Bulk views can also be disabled via the disable attribute:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
bulk_operations = ["create", "update", "delete"]
disable = ["bulk_update"] # Disable only bulk update
Valid disable values: bulk_create, bulk_update, bulk_delete.
Partial Success Semantics¶
Each item in a bulk request is processed independently. If an item fails (validation error, not found, etc.), the error is collected in the response but other items continue processing normally. This means:
- Successful items are committed to the database immediately.
- Failed items appear in
errors.detailswith the error message. - The response always returns HTTP
200with the combined result.
Performance
- Bulk Delete is optimized to use a single database query (
DELETE ... WHERE pk IN (...)) instead of deleting objects one by one. - Bulk Create and Bulk Update process items individually to preserve
save()validations and hooks.
Authentication
Bulk endpoints inherit per-verb authentication:
| Bulk Operation | Auth Source |
|---|---|
| Bulk Create | post_auth |
| Bulk Update | patch_auth |
| Bulk Delete | delete_auth |
Hooks and Validators
Bulk operations call the same hooks as single-item operations, per item:
parse_input_data()— field validation, FK resolution, base64 decodingcustom_actions()— custom field processing (create, update)post_create()— post-creation hook (create only)
Extra Decorators
Apply custom decorators to bulk endpoints via extra_decorators:
Core Attributes¶
| Attribute | Type | Default | Description |
|---|---|---|---|
bulk_operations |
list[Literal["create", "update", "delete"]] |
[] |
Which bulk operations to enable |
bulk_response_fields |
list[str] \| str \| None |
None |
Field(s) returned in success details (None = PK) |
bulk_create_docs |
str |
"Create multiple objects in a single request." |
Bulk create endpoint description |
bulk_update_docs |
str |
"Update multiple objects in a single request." |
Bulk update endpoint description |
bulk_delete_docs |
str |
"Delete multiple objects in a single request." |
Bulk delete endpoint description |
Ordering¶
Add native ordering support to the list endpoint with ordering_fields and default_ordering:
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
ordering_fields = ["created_at", "title", "views"]
default_ordering = "-created_at"
This automatically adds an ordering query parameter to the list endpoint:
# Single field ascending
GET /api/article/?ordering=title
# Single field descending
GET /api/article/?ordering=-title
# Multiple fields (comma-separated)
GET /api/article/?ordering=-views,title
# No ordering param → default_ordering applied
GET /api/article/
How It Works¶
- When
ordering_fieldsis set, anorderingquery parameter is automatically added to the filters schema. - The
orderingvalue is popped from the filters dict beforequery_params_handlerruns, so existing filter mixins never see it. - Each field in the comma-separated value is validated against
ordering_fields. Invalid fields are silently ignored. - Both
field(ascending) and-field(descending) are valid iffieldis inordering_fields. - If no valid fields remain,
default_orderingis applied as fallback. default_orderingaccepts a single string ("-created_at") or a list (["-created_at", "title"]).- When
ordering_fieldsis empty (default), the feature is completely disabled — no query parameter is added.
Ordering with Filters¶
Ordering works seamlessly with existing query_params and filter mixins:
@api.viewset(model=Article)
class ArticleViewSet(
APIViewSet,
IcontainsFilterViewSetMixin,
):
ordering_fields = ["title", "created_at"]
default_ordering = "-created_at"
query_params = {
"title": (str, None),
"is_published": (bool, None),
}
List Filtering¶
Define filters for the list view with query_params:
Override handler:
async def query_params_handler(self, queryset, filters: dict):
if filters.get("is_active") is not None:
queryset = queryset.filter(is_active=filters["is_active"])
if filters.get("role"):
queryset = queryset.filter(role=filters["role"])
if filters.get("search"):
from django.db.models import Q
s = filters["search"]
queryset = queryset.filter(Q(username__icontains=s) | Q(email__icontains=s))
return queryset
A dynamic Pydantic model (FiltersSchema) is built with pydantic.create_model from query_params.
List and Retrieve implementations¶
List now leverages ModelUtil.get_objects and list_read_s, automatically applying read optimizations and optional filters:
class ArticleViewSet(APIViewSet):
model = Article
api = api
def views(self):
@self.router.get("/")
async def list(request, filters: self.filters_schema = None):
qs = await self.model_util.get_objects(
request,
query_data=self._get_query_data(), # defaults from ModelSerializer.QuerySet.read
is_for_read=True,
)
if filters is not None:
qs = await self.query_params_handler(qs, filters.model_dump())
return await self.model_util.list_read_s(self.schema_out, request, qs)
Retrieve uses read_s with getters, deriving PK type from the model:
@self.router.get("/{pk}/")
async def retrieve(request, pk: self.path_schema):
return await self.model_util.read_s(
self.schema_out,
request,
query_data=QuerySchema(getters={"pk": self._get_pk(pk)}),
is_for_read=True,
)
- Path schema PK type is inferred from the model’s primary key field.
Many-to-Many Relations¶
Relations are declared via M2MRelationSchema objects (not tuples). Each schema can include:
model: related Django model or ModelSerializerrelated_name: attribute name on the main model (e.g."tags")path: custom URL segment (optional)auth: list of auth instances (optional)add: enable additions (bool)remove: enable removals (bool)get: enable GET listing (bool)filters: dict of{param_name: (type, default)}for relation-level filteringrelated_schema: optional pre-built schema for the related model (auto-generated if themodelis aModelSerializer)serializer_class: optionalSerializerclass for plain Django models. When provided,related_schemais auto-generated from the serializer. Cannot be used whenmodelis aModelSerializer.append_slash: bool to control trailing slash for the GET relation endpoint path. Defaults toFalse(no trailing slash) for backward compatibility. WhenTrue, the GET path ends with a trailing slash.verbose_name_plural: optional human-readable plural name for the related model, used in endpoint summaries and descriptions. When not provided, defaults to the related model's_meta.verbose_name_plural.get_decorators: optional list of decorators to apply to the GET (list related objects) endpoint. Decorators are applied viadecorate_view()alongside built-in decorators likeunique_viewandpaginate.post_decorators: optional list of decorators to apply to the POST (add/remove) endpoint. Decorators are applied viadecorate_view()alongside the built-inunique_viewdecorator.
If path is empty it falls back to the related model verbose name (lowercase plural).
If filters is provided, a per-relation filters schema is auto-generated and exposed on the GET relation endpoint:
GET /{base}/{pk}/{related_path}?param=value
Custom filter hook naming convention:
<related_name>_query_params_handler(self, queryset, filters_dict)
The M2M helper:
- Returns a paginated list of related items on GET.
- Supports both sync and async custom filter handlers.
- Uses
list_read_sfor related items serialization.
Example filter handler (sync or async):
def tags_query_params_handler(self, queryset, filters_dict):
name = filters_dict.get("name")
return queryset.filter(name=name) if name else queryset
# or
async def tags_query_params_handler(self, queryset, filters_dict):
# perform async lookups if needed, then return queryset
return queryset
Warning: Model support
- You can supply a standard Django
Model(not aModelSerializer) inM2MRelationSchema.model. When doing so you must provide eitherrelated_schemamanually orserializer_class:
from ninja_aio.models import serializers
class TagSerializer(serializers.Serializer):
class Meta:
model = Tag
schema_out = serializers.SchemaModelConfig(fields=["id", "name"])
M2MRelationSchema(
model=Tag, # plain django.db.models.Model
related_name="tags",
serializer_class=TagSerializer, # auto-generates related_schema
add=True,
remove=True,
get=True,
)
For ModelSerializer models, related_schema can be inferred automatically (via internal helpers).
Note: You cannot use serializer_class when model is already a ModelSerializer - this will raise a ValueError.
Example with filters:
class UserViewSet(APIViewSet):
model = User
api = api
m2m_relations = [
M2MRelationSchema(
model=Tag,
related_name="tags",
filters={"name": (str, "")}
)
]
async def tags_query_params_handler(self, queryset, filters):
name_filter = filters.get("name")
if name_filter:
queryset = queryset.filter(name__icontains=name_filter)
return queryset
Relation Handlers: GET filters vs POST per-PK resolution¶
-
GET filters handler (per relation):
-
Name:
<related_name>_query_params_handler(self, queryset, filters_dict) - Purpose: apply filters to the related list queryset (GET endpoint).
-
Supports both synchronous and asynchronous functions.
-
POST per-PK resolution handler (per relation):
- Name:
<related_name>_query_handler(self, request, pk, instance) - Purpose: resolve a single related object (for add/remove validation) before mutation.
- Must return a queryset; the object is resolved with
.afirst(). - Automatic fallback if missing:
ModelUtil(related_model).get_objects(request, ObjectsQuerySchema(filters={"pk": pk}))+.afirst().
Example:
class MyViewSet(APIViewSet):
model = Article
api = api
async def tags_query_params_handler(self, qs, filters: dict):
name = filters.get("name")
return qs.filter(name__icontains=name) if name else qs
async def tags_query_handler(self, request, pk, instance):
# allow only tags belonging to the same project as the instance
return Tag.objects.filter(pk=pk, project_id=instance.project_id)
Endpoint paths and operation naming¶
- GET relation:
/{base}/{pk}/{rel_path}by default (no trailing slash). You can enable a trailing slash per relation withappend_slash=True, resulting in/{base}/{pk}/{rel_path}/. - POST relation:
/{base}/{pk}/{rel_path}/(always with trailing slash).
Path normalization rules:
- Relation
pathis normalized internally; providingpathwith or without a leading slash produces the same final URL. - Example:
path="tags"orpath="/tags"both yieldGET /{base}/{pk}/tags(orGET /{base}/{pk}/tags/whenappend_slash=True) andPOST /{base}/{pk}/tags/. - If
pathis empty, it falls back to the related model verbose name.
Request/Response and concurrency¶
Request bodies:
- Add & Remove:
{ "add": number[], "remove": number[] } - Add only:
{ "add": number[] } - Remove only:
{ "remove": number[] }
Standard response (M2MSchemaOut):
- Concurrency:
aadd(...)andaremove(...)run in parallel viaasyncio.gatherwhen both lists are non-empty. - Per-PK errors include: object not found, state mismatch (removing non-related, adding already-related).
- Per-PK success messages indicate the executed action.
Generated M2M Endpoints (per relation)¶
| Method | Path | Feature |
|---|---|---|
| GET | /{base}/{pk}/{rel_path} |
List related objects (paginated, optional filters) |
| POST | /{base}/{pk}/{rel_path}/ |
Add/remove related objects |
Example:
class ArticleViewSet(APIViewSet):
model = Article
api = api
m2m_relations = [
M2MRelationSchema(model=Tag, related_name="tags"),
M2MRelationSchema(model=Category, related_name="categories", path="article-categories"),
M2MRelationSchema(model=User, related_name="authors", path="co-authors", auth=[AdminAuth()])
]
m2m_auth = [JWTAuth()] # fallback for relations without custom auth
Example with trailing slash on GET relation:
M2MRelationSchema(
model=Tag,
related_name="tags",
filters={"name": (str, "")},
append_slash=True, # GET /{base}/{pk}/tags/
)
Example with custom verbose name:
M2MRelationSchema(
model=Tag,
related_name="tags",
verbose_name_plural="Article Tags", # Used in summaries: "Get Article Tags", "Add or Remove Article Tags"
add=True,
remove=True,
get=True,
)
Example with custom decorators:
from functools import wraps
def cache_response(timeout=60):
"""Cache decorator for GET endpoints."""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# caching logic here
return await func(*args, **kwargs)
return wrapper
return decorator
def rate_limit(max_calls=100):
"""Rate limiting decorator for POST endpoints."""
def decorator(func):
@wraps(func)
async def wrapper(*args, **kwargs):
# rate limiting logic here
return await func(*args, **kwargs)
return wrapper
return decorator
M2MRelationSchema(
model=Tag,
related_name="tags",
get_decorators=[cache_response(timeout=300)], # Cache GET for 5 minutes
post_decorators=[rate_limit(max_calls=50)], # Rate limit POST operations
add=True,
remove=True,
get=True,
)
Note: Decorators are applied in addition to built-in decorators (unique_view, paginate). The order follows standard Python decorator stacking: decorators listed first are applied outermost.
Custom Views¶
Preferred (decorators): see the section above.
Legacy (still supported):
def views(self):
@self.router.get("/stats/", response={200: GenericMessageSchema})
async def stats(request):
total = await self.model.objects.acount()
return {"message": f"Total: {total}"}
Dynamic View Naming¶
All generated handlers are decorated with @unique_view(...) to ensure stable unique function names (prevents collisions and ensures consistent OpenAPI schema generation). Relation endpoints use explicit names like get_<model>_<rel_path> and manage_<model>_<rel_path>.
Extra Decorators¶
Apply custom decorators to specific CRUD operations via the extra_decorators attribute:
from ninja_aio.schemas.helpers import DecoratorsSchema
from functools import wraps
def log_operation(func):
@wraps(func)
async def wrapper(*args, **kwargs):
print(f"Calling {func.__name__}")
return await func(*args, **kwargs)
return wrapper
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
extra_decorators = DecoratorsSchema(
create=[log_operation],
update=[log_operation],
delete=[log_operation],
)
Available decorator fields:
- create: Decorators for create endpoint
- list: Decorators for list endpoint
- retrieve: Decorators for retrieve endpoint
- update: Decorators for update endpoint
- delete: Decorators for delete endpoint
- bulk_create: Decorators for bulk create endpoint
- bulk_update: Decorators for bulk update endpoint
- bulk_delete: Decorators for bulk delete endpoint
Overridable Hooks¶
| Hook | Purpose |
|---|---|
views() |
Register custom endpoints |
query_params_handler(queryset, filters) |
Apply list filters |
<related_name>_query_params_handler(queryset, filters) |
Apply relation-specific filters |
Error Handling¶
All CRUD and M2M endpoints may respond with GenericMessageSchema for error codes: 400 (validation), 401 (auth), 404 (not found).
Performance Tips¶
- Implement
@classmethod async def queryset_request(cls, request)in yourModelSerializerto prefetch related objects. - Use database indexes on filtered fields (
query_paramsand relationfilters). - Keep pagination enabled for large datasets.
- Prefetch reverse relations via
model_util.get_reverse_relations()(already applied in list view). - Limit slice size for expensive searches if needed (
queryset = queryset[:1000]).
Minimal Usage¶
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from .models import User
from ninja_aio.decorators import api_get
api = NinjaAIO(title="My API")
@api.viewset(model=User)
class UserViewSet(APIViewSet):
@api_get("/stats/")
async def stats(self, request):
total = await self.model.objects.acount()
return {"total": total}
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from .models import User
api = NinjaAIO(title="My API")
class UserViewSet(APIViewSet):
model = User
api = api
def views(self):
@self.router.get("/stats/")
async def stats(request):
total = await self.model.objects.acount()
return {"total": total}
UserViewSet().add_views_to_route()
Note: prefix and tags are optional. If omitted, the base path is inferred from the model verbose name plural and tags default to the model verbose name.
Disable Selected Views¶
@api.viewset(model=User)
class ReadOnlyUserViewSet(APIViewSet):
disable = ["create", "update", "delete"]
# Disable specific bulk operations
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
bulk_operations = ["create", "update", "delete"]
disable = ["bulk_delete"] # Enable bulk create/update, disable bulk delete
Authentication Example¶
@api.viewset(model=User)
class UserViewSet(APIViewSet):
auth = [JWTAuth()] # global fallback
get_auth = None # list/retrieve public
delete_auth = [AdminAuth()] # delete restricted
Complete M2M + Filters Example¶
Recommended:
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from ninja_aio.models import ModelSerializer
from ninja_aio.decorators import api_get
from django.db import models
api = NinjaAIO(title="My API")
class Tag(ModelSerializer):
name = models.CharField(max_length=100)
class ReadSerializer:
fields = ["id", "name"]
class User(ModelSerializer):
username = models.CharField(max_length=150)
tags = models.ManyToManyField(Tag, related_name="users")
class ReadSerializer:
fields = ["id", "username", "tags"]
@api.viewset(model=User)
class UserViewSet(APIViewSet):
query_params = {"search": (str, None)}
m2m_relations = [
M2MRelationSchema(
model=Tag,
related_name="tags",
filters={"name": (str, "")},
add=True,
remove=True,
get=True,
)
]
async def query_params_handler(self, queryset, filters):
if filters.get("search"):
from django.db.models import Q
s = filters["search"]
return queryset.filter(Q(username__icontains=s))
return queryset
async def tags_query_params_handler(self, queryset, filters):
name_filter = filters.get("name")
if name_filter:
queryset = queryset.filter(name__icontains=name_filter)
return queryset
Alternative implementation:
class UserViewSet(APIViewSet):
model = User
api = api
query_params = {"search": (str, None)}
m2m_relations = [
M2MRelationSchema(
model=Tag,
related_name="tags",
filters={"name": (str, "")},
add=True,
remove=True,
get=True,
)
]
async def query_params_handler(self, queryset, filters):
if filters.get("search"):
from django.db.models import Q
s = filters["search"]
return queryset.filter(Q(username__icontains=s))
return queryset
async def tags_query_params_handler(self, queryset, filters):
name_filter = filters.get("name")
if name_filter:
queryset = queryset.filter(name__icontains=name_filter)
return queryset
UserViewSet().add_views_to_route()
ReadOnlyViewSet¶
ReadOnlyViewSet enables only list and retrieve endpoints.
WriteOnlyViewSet¶
WriteOnlyViewSet enables only create, update, and delete endpoints.
See Also¶
-
ModelSerializer
-
Authentication
-
Pagination
-
APIView