Skip to content

Step 2 of 4

Create CRUD Views

Build a complete REST API with automatic CRUD operations using APIViewSet.

What You'll Learn

  • How to create a basic ViewSet
  • Understanding auto-generated endpoints
  • Customizing query parameters
  • Enabling bulk operations
  • Adding custom endpoints
  • Working with request context
  • Handling errors

Prerequisites — Make sure you've completed Step 1: Define Your Model. You should have the Article, Author, Category, and Tag models defined.


Basic ViewSet

Let's create a simple API for the Article model:

Python
# views.py
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from .models import Article

# Create API instance
api = NinjaAIO(  # (1)!
    title="Blog API",
    version="1.0.0",
    description="A simple blog API built with Django Ninja AIO"
)


@api.viewset(model=Article)  # (2)!
class ArticleViewSet(APIViewSet):
    pass  # (3)!
  1. NinjaAIO extends NinjaAPI with ORJSON rendering and built-in exception handling
  2. Registers the model and auto-generates all CRUD routes on the API router
  3. Zero configuration — endpoints, schemas, pagination, and OpenAPI docs are all automatic

That's it! You now have a complete CRUD API with 5 endpoints.

Configure URLs

Add the API to your Django URLs:

Python
# urls.py
from django.contrib import admin
from django.urls import path
from myapp.views import api

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", api.urls),
]

What Happens Under the Hood

graph LR
    A[Article Model] -->|"ReadSerializer<br/>CreateSerializer<br/>UpdateSerializer"| B[Pydantic Schemas]
    B --> C[APIViewSet]
    C -->|auto-generates| D["5 async endpoints<br/>+ OpenAPI docs<br/>+ pagination"]

    style A fill:#7c4dff,stroke:#7c4dff,color:#fff
    style B fill:#651fff,stroke:#651fff,color:#fff
    style C fill:#536dfe,stroke:#536dfe,color:#fff
    style D fill:#448aff,stroke:#448aff,color:#fff

Auto-Generated Endpoints

The ViewSet automatically generates these endpoints:

Method Endpoint Description Request Body Response
GET /api/article/ List all articles (paginated) None {count, next, previous, results}
POST /api/article/ Create new article Article data Created article
GET /api/article/{id} Retrieve single article None Article data
PATCH /api/article/{id}/ Update article Partial article data Updated article
DELETE /api/article/{id}/ Delete article None None (204)
Example error responses

404 — Object not found:

JSON
{"error": "article not found."}

400 — Validation error (missing required field):

JSON
{
  "detail": [
    {
      "type": "missing",
      "loc": ["body", "data", "title"],
      "msg": "Field required"
    }
  ]
}

401 — Unauthorized (when auth is configured):

JSON
{"detail": "Unauthorized"}

Test Your API

Start the development server:

Bash
python manage.py runserver

Visit http://localhost:8000/api/docs to see the auto-generated Swagger UI documentation.


Creating Multiple ViewSets

Let's add APIs for all our models:

Python
# views.py
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from .models import Article, Author, Category, Tag

api = NinjaAIO(title="Blog API", version="1.0.0")


@api.viewset(model=Author)
class AuthorViewSet(APIViewSet):
    pass


@api.viewset(model=Category)
class CategoryViewSet(APIViewSet):
    pass


@api.viewset(model=Tag)
class TagViewSet(APIViewSet):
    pass


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    pass

Now you have complete CRUD APIs for all models:

  • /api/author/
  • /api/category/
  • /api/tag/
  • /api/article/

Adding Query Parameters

Let's add filtering to the Article list endpoint:

Python
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    query_params = {  # (1)!
        "is_published": (bool, None),
        "author": (int, None),
        "category": (int, None),
        "search": (str, None),
    }

    async def query_params_handler(self, queryset, filters):  # (2)!
        # Filter by published status
        if filters.get("is_published") is not None:
            queryset = queryset.filter(is_published=filters["is_published"])

        # Filter by author
        if filters.get("author"):
            queryset = queryset.filter(author_id=filters["author"])

        # Filter by category
        if filters.get("category"):
            queryset = queryset.filter(category_id=filters["category"])

        # Search in title and content
        if filters.get("search"):
            from django.db.models import Q
            search_term = filters["search"]
            queryset = queryset.filter(
                Q(title__icontains=search_term) |
                Q(content__icontains=search_term)
            )

        return queryset  # (3)!
  1. Each entry is "param_name": (type, default) — a Pydantic schema is built at startup for OpenAPI docs and validation
  2. Called automatically on GET /api/article/?param=value — receives the filtered dict (only non-default values)
  3. Always return the queryset — the framework handles pagination and serialization

Using Query Parameters

Now you can filter articles:

Bash
# Get published articles
GET /api/article/?is_published=true

# Get articles by specific author
GET /api/article/?author=5

# Get articles in specific category
GET /api/article/?category=3

# Search articles
GET /api/article/?search=django

# Combine filters
GET /api/article/?is_published=true&author=5&category=3

# With pagination
GET /api/article/?is_published=true&page=2&page_size=20

Custom Endpoints (using @api_get / @api_post)

Add custom endpoints beyond CRUD using method decorators. For a higher-level alternative with automatic URL generation, auth inheritance, and detail/list distinction, see Custom Actions below.

Python
from ninja_aio.decorators import api_get, api_post


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    # Publish an article
    @api_post("/{pk}/publish/")
    async def publish(self, request, pk: int):
        article = await Article.objects.aget(pk=pk)
        article.is_published = True
        from django.utils import timezone
        article.published_at = timezone.now()
        await article.asave()

        return {
            "message": "Article published successfully",
            "published_at": article.published_at
        }

    # Unpublish an article
    @api_post("/{pk}/unpublish/")
    async def unpublish(self, request, pk: int):
        article = await Article.objects.aget(pk=pk)
        article.is_published = False
        article.published_at = None
        await article.asave()

        return {"message": "Article unpublished successfully"}

    # Increment view count
    @api_post("/{pk}/view/")
    async def increment_views(self, request, pk: int):
        article = await Article.objects.aget(pk=pk)
        article.views += 1
        await article.asave(update_fields=["views"])

        return {"views": article.views}

    # Get article statistics
    @api_get("/stats/")
    async def stats(self, request):
        from django.db.models import Count, Avg, Sum

        total = await Article.objects.acount()
        published = await Article.objects.filter(is_published=True).acount()

        # Use sync_to_async for aggregate
        from asgiref.sync import sync_to_async

        avg_views = await sync_to_async(
            lambda: Article.objects.aggregate(avg=Avg("views"))
        )()

        total_views = await sync_to_async(
            lambda: Article.objects.aggregate(total=Sum("views"))
        )()

        return {
            "total_articles": total,
            "published_articles": published,
            "draft_articles": total - published,
            "average_views": avg_views["avg"] or 0,
            "total_views": total_views["total"] or 0,
        }

    # Get popular articles
    @api_get("/popular/")
    async def popular(self, request, limit: int = 10):
        articles = []
        async for article in Article.objects.filter(
            is_published=True
        ).order_by("-views")[:limit]:
            articles.append(article)

        # Serialize articles
        from ninja_aio.models import ModelUtil
        util = ModelUtil(Article)
        schema = Article.generate_read_s()

        results = []
        for article in articles:
            data = await util.read_s(request, article, schema)
            results.append(data)

        return results

Custom Endpoint URLs

Your custom endpoints are now available:

Bash
# Publish article
POST /api/article/1/publish/

# Unpublish article
POST /api/article/1/unpublish/

# Increment views
POST /api/article/1/view/

# Get statistics
GET /api/article/stats/

# Get popular articles (top 10)
GET /api/article/popular/

# Get top 20
GET /api/article/popular/?limit=20

Custom Actions

Use the @action decorator to add custom endpoints to your ViewSet. Actions can operate on single instances (detail) or the collection (list):

Python
from ninja import Schema, Status
from ninja_aio.views import APIViewSet
from ninja_aio.decorators import action
from .models import Article


class CountSchema(Schema):
    count: int


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    # Detail action: operates on a single article
    @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"})

    # List action: operates on the collection
    @action(detail=False, methods=["get"], url_path="count", response=CountSchema)
    async def count(self, request):
        total = await self.model.objects.acount()
        return {"count": total}

    # Action with request body
    @action(detail=False, methods=["post"], url_path="import")
    async def import_articles(self, request, data: ArticleImportSchema):
        return Status(200, {"message": f"imported {len(data.items)} articles"})

    # Auto url_path from method name (underscores → hyphens)
    @action(detail=False, methods=["get"])
    async def recent_published(self, request):
        return {"message": "recent"}

This generates:

Method Endpoint Description
POST /api/article/{id}/activate Activate a single article
GET /api/article/count Count all articles
POST /api/article/import Import articles
GET /api/article/recent-published Recent published (auto path)

Key differences from @api_get / @api_post

Feature @action @api_get / @api_post
Detail actions (with pk) detail=True auto-adds {pk} Manual /{pk}/path
Multiple methods methods=["get", "post"] One decorator per method
Auth inheritance Inherits from viewset per-verb auth Manual auth=
URL path Auto-generated from method name Manual path required
OpenAPI metadata summary, description, tags, deprecated Same via kwargs

Tip

Actions are not affected by disable = ["all"] — they are always registered, even when all CRUD endpoints are disabled.


Request Context

Access request information in your ViewSet:

Python
from ninja_aio.decorators import api_get, api_post


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    @api_get("/my-articles/")
    async def my_articles(self, request):
        """Get articles by current user"""
        # Access authenticated user
        user = request.auth

        # Get user's articles
        articles = []
        async for article in Article.objects.filter(author=user):
            articles.append(article)

        # Serialize
        from ninja_aio.models import ModelUtil
        util = ModelUtil(Article)
        schema = Article.generate_read_s()

        results = []
        for article in articles:
            data = await util.read_s(request, article, schema)
            results.append(data)

        return results

    @api_post("/")
    async def create_article(self, request, data: Article.generate_create_s()):
        """Override create to set author from request"""
        # Set author from authenticated user
        data.author = request.auth.id

        # Use default create logic
        from ninja_aio.models import ModelUtil
        util = ModelUtil(Article)
        schema = Article.generate_read_s()

        return await util.create_s(request, data, schema)

Filtering by User

Automatically filter queryset based on user:

Python
class Article(ModelSerializer):
    # ... fields ...

    @classmethod
    async def queryset_request(cls, request):
        """Filter articles based on user"""
        qs = cls.objects.select_related('author', 'category').prefetch_related('tags')

        # If user is not authenticated, show only published
        if not request.auth:
            return qs.filter(is_published=True)

        # If user is admin, show all
        user = request.auth
        if user.is_staff:
            return qs

        # Regular users see published + their own drafts
        from django.db.models import Q
        return qs.filter(
            Q(is_published=True) | Q(author=user)
        )


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    pass
    # queryset_request is automatically called for all operations

Custom Pagination

Override default pagination:

Python
from ninja.pagination import PageNumberPagination


class LargePagePagination(PageNumberPagination):
    page_size = 50  # Default 50 items per page
    max_page_size = 200  # Allow up to 200 items


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    pagination_class = LargePagePagination

Now list endpoint uses custom pagination:

Bash
# Default 50 items
GET /api/article/

# Custom page size
GET /api/article/?page_size=100

# Page 2
GET /api/article/?page=2&page_size=50

Ordering

Add native ordering to the list endpoint with ordering_fields and default_ordering:

Python
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    ordering_fields = ["created_at", "title", "views", "published_at"]
    default_ordering = "-created_at"  # Newest first by default

That's it! The framework automatically adds an ordering query parameter to the list endpoint and validates fields.

Usage:

Bash
# Newest first (default)
GET /api/article/

# Oldest first
GET /api/article/?ordering=created_at

# By title A-Z
GET /api/article/?ordering=title

# By title Z-A
GET /api/article/?ordering=-title

# Multiple fields (comma-separated)
GET /api/article/?ordering=-views,title

# Combined with filters
GET /api/article/?ordering=-created_at&is_published=true

Tip

Invalid field names are silently ignored. If all requested fields are invalid, default_ordering is applied.


Error Handling

Handle errors gracefully:

Python
from ninja_aio.exceptions import SerializeError, NotFoundError
from ninja_aio.decorators import api_post


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    @api_post("/{pk}/publish/")
    async def publish(self, request, pk: int):
        try:
            article = await Article.objects.aget(pk=pk)
        except Article.DoesNotExist:
            raise NotFoundError(self.model)

        # Check if already published
        if article.is_published:
            raise SerializeError(
                {"article": "already published"},
                status_code=400
            )

        # Publish
        article.is_published = True
        from django.utils import timezone
        article.published_at = timezone.now()
        await article.asave()

        return {
            "message": "Article published successfully",
            "published_at": article.published_at
        }

Bulk Operations

Need to create, update, or delete multiple objects at once? Enable bulk operations:

Python
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    bulk_operations = ["create", "update", "delete"]

This adds three new endpoints to your API:

Method Endpoint Description Request Body Response
POST /api/articles/bulk/ Create multiple articles [{...}, {...}] 200 BulkResultSchema
PATCH /api/articles/bulk/ Update multiple articles [{id, ...}, {id, ...}] 200 BulkResultSchema
DELETE /api/articles/bulk/ Delete multiple articles {"ids": [1, 2, 3]} 200 BulkResultSchema

Bulk operations use partial success semantics — each item is processed independently. Successful items are committed while failures are collected in the response without affecting other items.

Response Format

All bulk endpoints return a BulkResultSchema:

JSON
{
  "success": {
    "count": 2,
    "details": [1, 3]
  },
  "errors": {
    "count": 1,
    "details": [{"error": "Not found."}]
  }
}
  • success.details contains the primary keys of successfully processed objects (default). Customizable via bulk_response_fields.
  • errors.details contains error details for each failed item.

Bulk Create

Send a list of objects to create them all in one request:

Bash
curl -X POST http://localhost:8000/api/article/bulk/ \
  -H "Content-Type: application/json" \
  -d '[
    {"title": "Article 1", "content": "Content 1", "author": 1},
    {"title": "Article 2", "content": "Content 2", "author": 1},
    {"title": "Article 3", "content": "Content 3", "author": 2}
  ]'

Bulk Update

Send a list of objects with their primary key and the fields to update:

Bash
curl -X PATCH http://localhost:8000/api/article/bulk/ \
  -H "Content-Type: application/json" \
  -d '[
    {"id": 1, "title": "Updated Title 1"},
    {"id": 2, "title": "Updated Title 2", "is_published": true}
  ]'

Bulk Delete

Send a list of primary keys to delete. This operation is optimized — it uses a single database query to delete all existing objects instead of deleting them one by one:

Bash
curl -X DELETE http://localhost:8000/api/article/bulk/ \
  -H "Content-Type: application/json" \
  -d '{"ids": [1, 2, 3]}'

Custom Response Fields

By default, success.details returns primary keys. Use bulk_response_fields to customize what's returned:

Python
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    bulk_operations = ["create", "update", "delete"]
    bulk_response_fields = "title"  # Returns ["Article 1", "Article 2"]
    # Or return multiple fields as dicts:
    # bulk_response_fields = ["id", "title"]  # Returns [{"id": 1, "title": "..."}, ...]

Selective Bulk Operations

You can enable only the operations you need:

Python
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    bulk_operations = ["create"]  # Only bulk create

Tip

Bulk operations reuse your existing schema_in and schema_update schemas, so all validations and hooks (like custom_actions and post_create) are applied per item.


Partial Update Validation

By default, PATCH endpoints accept empty payloads. Enable validation to reject them:

Python
@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    require_update_fields = True

Empty PATCH requests will return a 400 error: "No fields provided for update.". This also applies to bulk updates — empty items are collected as errors in partial success responses.


Disabling Endpoints

Disable specific CRUD operations:

Python
@api.viewset(model=Category)
class CategoryViewSet(APIViewSet):
    # Disable delete (categories can't be deleted)
    disable_delete = True

    # Disable update (categories are immutable)
    disable_update = True

Now only these endpoints are available:

  • GET /api/category/ - List
  • POST /api/category/ - Create
  • GET /api/category/{id} - Retrieve

Response Customization

Customize response format:

Python
from ninja_aio.decorators import api_get


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    @api_get("/{pk}/")
    async def retrieve(self, request, pk: int):
        """Custom retrieve with additional data"""
        article = await Article.objects.select_related(
            'author', 'category'
        ).prefetch_related('tags').aget(pk=pk)

        # Serialize article
        from ninja_aio.models import ModelUtil
        util = ModelUtil(Article)
        schema = Article.generate_read_s()
        article_data = await util.read_s(request, article, schema)

        # Get related articles
        related = []
        async for rel_article in Article.objects.filter(
            category=article.category,
            is_published=True
        ).exclude(pk=pk)[:5]:
            rel_data = await util.read_s(request, rel_article, schema)
            related.append(rel_data)

        # Get author's other articles
        author_articles = []
        async for auth_article in Article.objects.filter(
            author=article.author,
            is_published=True
        ).exclude(pk=pk)[:5]:
            auth_data = await util.read_s(request, auth_article, schema)
            author_articles.append(auth_data)

        return {
            "article": article_data,
            "related_articles": related,
            "author_articles": author_articles,
            "meta": {
                "total_views": article.views,
                "author_article_count": await Article.objects.filter(
                    author=article.author
                ).acount()
            }
        }

Complete Example

Here's a complete ViewSet with all features:

Full ViewSet code (click to expand)
Python
# views.py
from ninja_aio import NinjaAIO
from ninja_aio.views import APIViewSet
from ninja.pagination import PageNumberPagination
from ninja_aio.exceptions import SerializeError, NotFoundError
from ninja_aio.decorators import api_get, api_post
from .models import Article, Author, Category, Tag
from django.db.models import Q

api = NinjaAIO(
    title="Blog API",
    version="1.0.0",
    description="A complete blog API"
)


class CustomPagination(PageNumberPagination):
    page_size = 20
    max_page_size = 100


@api.viewset(model=Author)
class AuthorViewSet(APIViewSet):
    pass


@api.viewset(model=Category)
class CategoryViewSet(APIViewSet):
    pass


@api.viewset(model=Tag)
class TagViewSet(APIViewSet):
    pass


@api.viewset(model=Article)
class ArticleViewSet(APIViewSet):
    pagination_class = CustomPagination

    query_params = {
        "is_published": (bool, None),
        "author": (int, None),
        "category": (int, None),
        "tag": (int, None),
        "search": (str, None),
        "ordering": (str, "-created_at"),
    }

    async def query_params_handler(self, queryset, filters):
        # Published filter
        if filters.get("is_published") is not None:
            queryset = queryset.filter(is_published=filters["is_published"])

        # Author filter
        if filters.get("author"):
            queryset = queryset.filter(author_id=filters["author"])

        # Category filter
        if filters.get("category"):
            queryset = queryset.filter(category_id=filters["category"])

        # Tag filter
        if filters.get("tag"):
            queryset = queryset.filter(tags__id=filters["tag"])

        # Search
        if filters.get("search"):
            search = filters["search"]
            queryset = queryset.filter(
                Q(title__icontains=search) |
                Q(content__icontains=search) |
                Q(excerpt__icontains=search)
            )

        # Ordering
        ordering = filters.get("ordering", "-created_at")
        valid_orderings = [
            "created_at", "-created_at",
            "title", "-title",
            "views", "-views",
            "published_at", "-published_at"
        ]
        if ordering in valid_orderings:
            queryset = queryset.order_by(ordering)

        return queryset

    # Publish article
    @api_post("/{pk}/publish/")
    async def publish(self, request, pk: int):
        try:
            article = await Article.objects.aget(pk=pk)
        except Article.DoesNotExist:
            raise NotFoundError(self.model)

        if article.is_published:
            raise SerializeError(
                {"article": "already published"},
                status_code=400
            )

        article.is_published = True
        from django.utils import timezone
        article.published_at = timezone.now()
        await article.asave()

        return {"message": "Article published", "published_at": article.published_at}

    # Unpublish article
    @api_post("/{pk}/unpublish/")
    async def unpublish(self, request, pk: int):
        try:
            article = await Article.objects.aget(pk=pk)
        except Article.DoesNotExist:
            raise NotFoundError(self.model)

        article.is_published = False
        article.published_at = None
        await article.asave()

        return {"message": "Article unpublished"}

    # Increment views
    @api_post("/{pk}/view/")
    async def view(self, request, pk: int):
        try:
            article = await Article.objects.aget(pk=pk)
        except Article.DoesNotExist:
            raise NotFoundError(self.model)

        article.views += 1
        await article.asave(update_fields=["views"])

        return {"views": article.views}

    # Statistics
    @api_get("/stats/")
    async def stats(self, request):
        from django.db.models import Count, Avg, Sum
        from asgiref.sync import sync_to_async

        total = await Article.objects.acount()
        published = await Article.objects.filter(is_published=True).acount()

        avg_views = await sync_to_async(
            lambda: Article.objects.aggregate(avg=Avg("views"))
        )()

        total_views = await sync_to_async(
            lambda: Article.objects.aggregate(total=Sum("views"))
        )()

        return {
            "total_articles": total,
            "published": published,
            "drafts": total - published,
            "avg_views": avg_views["avg"] or 0,
            "total_views": total_views["total"] or 0,
        }

    # Popular articles
    @api_get("/popular/")
    async def popular(self, request, limit: int = 10):
        articles = []
        async for article in Article.objects.filter(
            is_published=True
        ).order_by("-views")[:limit]:
            articles.append(article)

        from ninja_aio.models import ModelUtil
        util = ModelUtil(Article)
        schema = Article.generate_read_s()

        results = []
        for article in articles:
            data = await util.read_s(request, article, schema)
            results.append(data)

        return results

Testing Your API

Test your endpoints using curl, httpie, or the Swagger UI:

Bash
# List articles
curl http://localhost:8000/api/article/

# Create article
curl -X POST http://localhost:8000/api/article/ \
  -H "Content-Type: application/json" \
  -d '{
    "title": "My Article",
    "content": "Content here...",
    "author": 1,
    "category": 2
  }'

# Get article
curl http://localhost:8000/api/article/1

# Update article
curl -X PATCH http://localhost:8000/api/article/1/ \
  -H "Content-Type: application/json" \
  -d '{"title": "Updated Title"}'

# Delete article
curl -X DELETE http://localhost:8000/api/article/1/

# Custom endpoints
curl -X POST http://localhost:8000/api/article/1/publish/
curl http://localhost:8000/api/article/stats/
curl http://localhost:8000/api/article/popular/?limit=5

Ready for the next step?

Now that you have CRUD operations set up, let's add authentication!

Step 3: Add Authentication

What You've Learned

  • Creating ViewSets for CRUD operations
  • Understanding auto-generated endpoints
  • Adding query parameters and filtering
  • Creating custom endpoints
  • Working with pagination
  • Handling errors properly
  • Using bulk operations
  • Customizing responses