Django Ninja Aio Crud

Async CRUD framework for Django Ninja Automatic schema generation · Filtering · Pagination · Auth · M2M management

Documentation · PyPI · Framework Comparison · Performance Benchmarks · Example Project · Issues

---

Features

| | Feature | Description | |---|---|---| | 🔒 Type Safety | Generic classes | Full IDE autocomplete and type checking with generic ModelUtil, Serializer, and APIViewSet | | Meta-driven Serializer | Dynamic schemas | Generate CRUD schemas for existing Django models without changing base classes | | Async CRUD ViewSets | Full operations | Create, list, retrieve, update, delete — all async | | Auto Schemas | Pydantic generation | Automatic read/create/update schemas from ModelSerializer | | Dynamic Query Params | Runtime schemas | Built with pydantic.create_model for flexible filtering | | Per-method Auth | Granular control | auth, get_auth, post_auth, etc. | | Async Pagination | Customizable | PageNumberPagination, CursorPagination, or custom — DB-level slicing | | M2M Relations | Add/remove/list | Endpoints via M2MRelationSchema with filtering support | | Reverse Relations | Nested serialization | Automatic handling of reverse FK and M2M | | Bulk Operations | Create/update/delete | Opt-in bulk endpoints with partial success semantics and configurable response fields | | Custom Actions | @action decorator | Detail and list actions with auth inheritance, custom decorators, and auto URL generation | | Lifecycle Hooks | Extensible | before_save, after_save, custom_actions, on_delete, and more | | Schema Validators | Pydantic validators | @field_validator and @model_validator on serializer classes | | ORJSON Renderer | Performance | Built-in fast JSON rendering via NinjaAIO | | AI Agent Integration | MCP tools | Expose ViewSets as MCP tools for any MCP client |

---

See It In Action

A ModelSerializer-based model, wired to a viewset, serving full CRUD in a few lines — no manual schemas or endpoint wiring. See the docs for the full walkthrough.

---

Quick Start

Option A: Meta-driven Serializer (existing models)

Use this if you already have Django models and don't want to change their base class.

from ninja_aio.models import serializers
from ninja_aio.views import APIViewSet
from ninja_aio import NinjaAIO
from . import models

class BookSerializer(serializers.Serializer):
    class Meta:
        model = models.Book
        schema_in = serializers.SchemaModelConfig(fields=["title", "published"])
        schema_out = serializers.SchemaModelConfig(fields=["id", "title", "published"])
        schema_update = serializers.SchemaModelConfig(
            optionals=[("title", str), ("published", bool)]
        )

api = NinjaAIO()

@api.viewset(models.Book)
class BookViewSet(APIViewSet):
    serializer_class = BookSerializer

Option B: ModelSerializer (new projects)

Define models with built-in serialization for minimal boilerplate.

models.py

from django.db import models
from ninja_aio.models import ModelSerializer

class Book(ModelSerializer):
    title = models.CharField(max_length=120)
    published = models.BooleanField(default=True)

    class ReadSerializer:
        fields = ["id", "title", "published"]

    class CreateSerializer:
        fields = ["title", "published"]

    class UpdateSerializer:
        optionals = [("title", str), ("published", bool)]

views.py

from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from .models import Book

api = NinjaAIO()

@api.viewset(Book)
class BookViewSet(APIViewSet):
    pass

> Visit /docs — CRUD endpoints ready.

---

Query Filtering

@api.viewset(Book)
class BookViewSet(APIViewSet):
    query_params = {"published": (bool, None), "title": (str, None)}

    async def query_params_handler(self, queryset, filters):
        if filters.get("published") is not None:
            queryset = queryset.filter(published=filters["published"])
        if filters.get("title"):
            queryset = queryset.filter(title__icontains=filters["title"])
        return queryset

GET /book/?published=true&title=python

---

Many-to-Many Relations

from ninja_aio.schemas import M2MRelationSchema

class Tag(ModelSerializer):
    name = models.CharField(max_length=50)
    class ReadSerializer:
        fields = ["id", "name"]

class Article(ModelSerializer):
    title = models.CharField(max_length=120)
    tags = models.ManyToManyField(Tag, related_name="articles")
    class ReadSerializer:
        fields = ["id", "title", "tags"]

@api.viewset(Article)
class ArticleViewSet(APIViewSet):
    m2m_relations = [
        M2MRelationSchema(
            model=Tag,
            related_name="tags",
            filters={"name": (str, "")}
        )
    ]

    async def tags_query_params_handler(self, queryset, filters):
        n = filters.get("name")
        if n:
            queryset = queryset.filter(name__icontains=n)
        return queryset

Endpoints:

GET  /article/{pk}/tag?name=dev
POST /article/{pk}/tag/    body: {"add": [1, 2], "remove": [3]}

---

Authentication (JWT)

from ninja_aio.auth import AsyncJwtBearer
from joserfc import jwk

class JWTAuth(AsyncJwtBearer):
    jwt_public = jwk.RSAKey.import_key("-----BEGIN PUBLIC KEY----- ...")
    jwt_alg = "RS256"
    claims = {"sub": {"essential": True}}

    async def auth_handler(self, request):
        book_id = self.dcd.claims.get("sub")
        return await Book.objects.aget(id=book_id)

@api.viewset(Book)
class SecureBookViewSet(APIViewSet):
    auth = [JWTAuth()]
    get_auth = None  # list/retrieve remain public

---

Lifecycle Hooks

Available on every save/delete cycle:

| Hook | When | |---|---| | on_create_before_save | Before first save | | on_create_after_save | After first save | | before_save | Before any save | | after_save | After any save | | on_delete | After deletion | | custom_actions(payload) | Create/update custom field logic | | post_create() | After create commit |

---

Custom Endpoints

Option A: @action Decorator (recommended)

from ninja import Schema, Status
from ninja_aio.decorators import action

class StatsSchema(Schema):
    total: int

@api.viewset(Book)
class BookViewSet(APIViewSet):
    @action(detail=False, methods=["get"], url_path="stats", response=StatsSchema)
    async def stats(self, request):
        total = await Book.objects.acount()
        return {"total": total}

    @action(detail=True, methods=["post"], url_path="publish")
    async def publish(self, request, pk):
        book = await self.model_util.get_object(request, pk)
        book.published = True
        await book.asave()
        return Status(200, {"message": "published"})

GET  /book/stats/       → {"total": 42}
POST /book/{pk}/publish/ → {"message": "published"}

Option B: operations Decorators

from ninja_aio.decorators import api_get

@api.viewset(Book)
class BookViewSet(APIViewSet):
    @api_get("/stats/")
    async def stats(self, request):
        total = await Book.objects.acount()
        return {"total": total}

---

AI Agent Integration (MCP)

Expose every registered APIViewSet — CRUD, bulk operations, and custom @action/@on endpoints — as well as any custom APIView, as MCP tools any MCP client can call directly.

pip install "django-ninja-aio-crud[mcp]"

Option A: manage.py mcp_server (recommended)

Add "ninja_aio" to INSTALLED_APPS to pick up the bundled management command:

INSTALLED_APPS = [
    ...,
    "ninja_aio",
]

python manage.py mcp_server myproject.api.api

Or set a default so you can drop the argument:

# settings.py
NINJA_AIO_MCP_API = "myproject.api.api"

{
  "mcpServers": {
    "myproject": {
      "type": "stdio",
      "command": "python",
      "args": ["manage.py", "mcp_server"]
    }
  }
}

Option B: standalone script

# mcp_server.py
import asyncio
import django
django.setup()

from myproject.api import api  # your NinjaAIO() instance with @api.viewset(...) registered
from ninja_aio.mcp import run_mcp_server

if __name__ == "__main__":
    asyncio.run(run_mcp_server(api))

{
  "mcpServers": {
    "myproject": {
      "type": "stdio",
      "command": "python",
      "args": ["mcp_server.py"]
    }
  }
}

Every @api.viewset(...)-registered ViewSet and @api.view(...)-registered View is picked up automatically (or pass viewsets=[...]/views=[...] explicitly). Tools are named _ for ViewSets — e.g. book_create, book_list, book_retrieve, book_update, book_delete, book_bulk_create, book_publish — and __ for plain Views — e.g. bookview_stats_get.

> ⚠️ Auth caveat: tool calls invoke the same registered view logic as HTTP requests (filters, pagination, and on_before_operation/on_before_object_operation/query_params_handler hooks all run identically) but bypass django-ninja's auth= wiring, since that applies at the router layer, not inside the handler. Pass request_factory to attach your own request.user/auth context, and use viewset hooks to enforce authorization for MCP-driven calls:

from ninja_aio.mcp import NinjaAIOMCPServer

def mcp_request_factory():
    from django.test.client import AsyncRequestFactory
    request = AsyncRequestFactory().get("/mcp/")
    request.user = get_service_account_user()  # your own resolution logic
    return request

server = NinjaAIOMCPServer(api, request_factory=mcp_request_factory)

---

Bulk Operations

@api.viewset(Book)
class BookViewSet(APIViewSet):
    bulk_operations = ["create", "update", "delete"]
    bulk_response_fields = "title"  # Optional: return titles instead of PKs

POST   /book/bulk/  body: [{...}, {...}]         → {"success": {"count": 2, "details": ["Book 1", "Book 2"]}}
PATCH  /book/bulk/  body: [{id, ...}, {id, ...}] → {"success": {"count": 2, "details": ["Updated 1", "Updated 2"]}}
DELETE /book/bulk/  body: {"ids": [1, 2]}         → {"success": {"count": 2, "details": ["Book 1", "Book 2"]}}

---

Pagination

Default: PageNumberPagination. Override per ViewSet:

from ninja.pagination import PageNumberPagination, CursorPagination

class LargePagination(PageNumberPagination):
    page_size = 50
    max_page_size = 200

@api.viewset(Book)
class BookViewSet(APIViewSet):
    pagination_class = LargePagination
    # Or use cursor-based pagination for large datasets:
    # pagination_class = CursorPagination

---

Schema Validators

Add Pydantic @field_validator and @model_validator directly on serializer classes for input validation.

ModelSerializer

Declare validators on inner serializer classes:

from django.db import models
from pydantic import field_validator, model_validator
from ninja_aio.models import ModelSerializer

class Book(ModelSerializer):
    title = models.CharField(max_length=120)
    description = models.TextField(blank=True)

    class CreateSerializer:
        fields = ["title", "description"]

        @field_validator("title")
        @classmethod
        def validate_title_min_length(cls, v):
            if len(v) 

---

## License

MIT License. See [LICENSE](LICENSE).

## Source & license

This open-source MCP server is cataloged on AgentStack and links to its original source — we do not rehost the code.

- **Author:** [caspel26](https://github.com/caspel26)
- **Source:** [caspel26/django-ninja-aio-crud](https://github.com/caspel26/django-ninja-aio-crud)
- **License:** MIT
- **Homepage:** https://django-ninja-aio.com

Install and usage instructions live in the source repository linked above.