django-ninja · diff
v1.1.0 to v2.0.0
10 added, 1392 removed. Audit A to A.
---
name: django-ninja
- description: "Fast Django REST framework with Pydantic validation, type hints, and automatic OpenAPI documentation for building type-safe APIs"
+ description: Use when building Django REST APIs with django-ninja - Pydantic schemas, routers, CRUD endpoints, authentication, pagination, file uploads, async views, OpenAPI docs, or migrating from DRF
metadata:
author: mte90
- version: "1.1.0"
+ version: 2.0.0
tags:
- python
- django
- rest-api
- pydantic
- openapi
- type-safe
---
# Django Ninja
Complete reference for building fast, type-safe REST APIs with Django and Pydantic.
## Overview
Django Ninja is a web framework for building APIs with Django and Python 3.6+ type hints.
**Versions**: django-ninja 1.6.x + Django 6.0 compatible. Requires Pydantic v2. It provides automatic request validation, response serialization, and generates OpenAPI documentation.
**Key Features:**
- Fast: Built on Pydantic for high performance
- Type-safe: Full IDE autocomplete and type checking
- Auto docs: Automatic OpenAPI/Swagger documentation
- Easy: Django integration with minimal boilerplate
- Async support: Native async/await support
### Django Ninja vs Django REST Framework
| Feature | Django Ninja | DRF |
|---------|-------------|-----|
| Validation | Pydantic | Serializers |
| Performance | Very Fast | Fast |
| Type Safety | Full | Partial |
| OpenAPI | Auto | Manual (drf-spectacular) |
| Learning Curve | Easy | Moderate |
| Async | Native | Limited |
**Use Django Ninja when:**
- Building new REST APIs
- Need type safety and IDE support
- Want automatic OpenAPI docs
- Prefer Pydantic validation
**Use DRF when:**
- Have existing DRF codebase
- Need browsable API (HTML responses)
- Require specific DRF features
## Installation
```bash
pip install django-ninja
```
### Django Integration
```python
# settings.py
INSTALLED_APPS = [
# ...
'ninja',
]
```
### Basic Setup
```python
# api.py
from ninja import NinjaAPI
api = NinjaAPI()
@api.get("/hello")
def hello(request):
return {"message": "Hello World"}
# urls.py
from django.urls import path
from .api import api
urlpatterns = [
path("api/", api.urls),
]
```
### Project Structure
```
myproject/
├── api/
│ ├── __init__.py
│ ├── urls.py
│ ├── schemas.py
│ ├── views/
│ │ ├── __init__.py
│ │ ├── users.py
│ │ └── posts.py
│ └── auth.py
├── models.py
└── settings.py
```
## Schema Definitions
### Pydantic v2 Context Support
```python
# Django 6.0+ context access in schemas
class Payload(Schema):
id: int
request_path: str
@staticmethod
def resolve_request_path(data, context):
return context["request"].get_full_path()
```
### Basic Schema
```python
from ninja import Schema
from datetime import datetime
from typing import Optional, List
class UserIn(Schema):
username: str
email: str
password: str
first_name: Optional[str] = None
last_name: Optional[str] = None
class UserOut(Schema):
id: int
username: str
email: str
first_name: Optional[str] = None
last_name: Optional[str] = None
created_at: datetime
class UserUpdate(Schema):
username: Optional[str] = None
email: Optional[str] = None
first_name: Optional[str] = None
last_name: Optional[str] = None
```
### ModelSchema (from Django Models)
```python
from ninja import ModelSchema
from .models import User, Post
class UserSchema(ModelSchema):
class Config:
model = User
model_fields = ['id', 'username', 'email', 'first_name', 'last_name']
class PostSchema(ModelSchema):
author: UserSchema # Nested schema
class Config:
model = Post
model_fields = ['id', 'title', 'slug', 'body', 'publish', 'status']
class PostCreateSchema(ModelSchema):
class Config:
model = Post
model_fields = ['title', 'body', 'status']
model_fields_optional = ['status'] # Optional fields
class PostUpdateSchema(ModelSchema):
class Config:
model = Post
model_fields = ['title', 'body', 'status']
model_fields_optional = '__all__' # All fields optional
```
### Nested Schemas
```python
from typing import List
class CommentSchema(Schema):
id: int
content: str
author: str
created_at: datetime
class PostDetailSchema(Schema):
id: int
title: str
slug: str
body: str
author: UserSchema
categories: List[str]
comments: List[CommentSchema]
created_at: datetime
updated_at: datetime
```
### Validators
```python
from ninja import Schema
from pydantic import validator, root_validator
import re
class UserCreate(Schema):
username: str
email: str
password: str
@validator('username')
def validate_username(cls, v):
if len(v) < 3:
raise ValueError('Username must be at least 3 characters')
if not re.match(r'^[a-zA-Z0-9_]+$', v):
raise ValueError('Username can only contain letters, numbers, and underscores')
return v
@validator('email')
def validate_email(cls, v):
if '@' not in v:
raise ValueError('Invalid email format')
return v.lower()
@validator('password')
def validate_password(cls, v):
if len(v) < 8:
raise ValueError('Password must be at least 8 characters')
return v
@root_validator
def validate_all(cls, values):
# Cross-field validation
if values.get('username') == values.get('password'):
raise ValueError('Password cannot be the same as username')
return values
```
### Custom Types
```python
from ninja import Schema
from typing import Annotated, List
from pydantic import Field, HttpUrl
# Using Annotated for reusable validators
class PostCreate(Schema):
title: Annotated[str, Field(min_length=5, max_length=200)]
slug: Annotated[str, Field(pattern=r'^[a-z0-9-]+$')]
body: Annotated[str, Field(min_length=10)]
tags: Annotated[List[str], Field(max_items=10)] = []
# Custom type with validator
class URLSchema(Schema):
url: HttpUrl # Validates URL format
# Using constrint
from pydantic import constr
ShortStr = constr(max_length=50)
EmailStr = constr(regex=r'^[^@]+@[^@]+\.[^@]+$')
class ContactSchema(Schema):
name: ShortStr
email: EmailStr
```
## Router & API
### HTTP Methods
```python
from ninja import NinjaAPI
from .schemas import UserIn, UserOut, PostSchema
api = NinjaAPI()
# GET - Retrieve resources
@api.get("/users", response=List[UserOut])
def list_users(request):
return User.objects.all()
@api.get("/users/{user_id}", response=UserOut)
def get_user(request, user_id: int):
user = get_object_or_404(User, id=user_id)
return user
# POST - Create resources
@api.post("/users", response=UserOut)
def create_user(request, payload: UserIn):
user = User.objects.create_user(**payload.dict())
return user
# PUT - Full update
@api.put("/users/{user_id}", response=UserOut)
def update_user(request, user_id: int, payload: UserUpdate):
user = get_object_or_404(User, id=user_id)
for attr, value in payload.dict(exclude_unset=True).items():
setattr(user, attr, value)
user.save()
return user
# PATCH - Partial update
@api.patch("/users/{user_id}", response=UserOut)
def partial_update_user(request, user_id: int, payload: UserUpdate):
user = get_object_or_404(User, id=user_id)
for attr, value in payload.dict(exclude_unset=True).items():
setattr(user, attr, value)
user.save()
return user
# DELETE
@api.delete("/users/{user_id}")
def delete_user(request, user_id: int):
user = get_object_or_404(User, id=user_id)
user.delete()
return {"success": True}
```
### Path Parameters
```python
from ninja import Path
@api.get("/posts/{post_id}/comments/{comment_id}")
def get_comment(
request,
post_id: int,
comment_id: int
):
"""Path parameters with type hints are automatically validated."""
comment = get_object_or_404(Comment, id=comment_id, post_id=post_id)
return {"comment": comment.content}
# String path parameters
@api.get("/categories/{slug}/posts")
def category_posts(request, slug: str):
category = get_object_or_404(Category, slug=slug)
return category.posts.all()
# UUID path parameters
import uuid
@api.get("/orders/{order_id}")
def get_order(request, order_id: uuid.UUID):
order = get_object_or_404(Order, id=order_id)
return order
```
### Query Parameters
```python
from ninja import Query, Schema
from typing import Optional, List
from datetime import date
class FilterParams(Schema):
search: Optional[str] = None
status: Optional[str] = None
category: Optional[int] = None
tags: Optional[List[str]] = None
created_after: Optional[date] = None
created_before: Optional[date] = None
ordering: Optional[str] = "-created_at"
page: int = 1
page_size: int = 20
@api.get("/posts", response=List[PostSchema])
def list_posts(request, filters: FilterParams = Query(...)):
"""Query parameters are automatically validated and parsed."""
posts = Post.objects.all()
if filters.search:
posts = posts.filter(
Q(title__icontains=filters.search) |
Q(body__icontains=filters.search)
)
if filters.status:
posts = posts.filter(status=filters.status)
if filters.category:
posts = posts.filter(categories__id=filters.category)
if filters.tags:
posts = posts.filter(tags__name__in=filters.tags)
if filters.created_after:
posts = posts.filter(created_at__gte=filters.created_after)
if filters.created_before:
posts = posts.filter(created_at__lte=filters.created_before)
posts = posts.order_by(filters.ordering)
return posts[(filters.page - 1) * filters.page_size:filters.page * filters.page_size]
```
### Request Body
```python
from ninja import Body, Schema
from typing import List
class PostCreate(Schema):
title: str
body: str
category_ids: List[int]
tags: List[str] = []
@api.post("/posts", response=PostSchema)
def create_post(request, payload: PostCreate):
"""Request body is automatically validated against schema."""
# payload is a Pydantic model instance
post = Post.objects.create(
title=payload.title,
body=payload.body,
author=request.user
)
if payload.category_ids:
post.categories.set(payload.category_ids)
if payload.tags:
post.tags.set(payload.tags)
return post
# Multiple body parameters
@api.post("/posts/{post_id}/comments")
def add_comment(
request,
post_id: int,
content: str = Body(...),
author_name: str = Body(...),
author_email: str = Body(...)
):
post = get_object_or_404(Post, id=post_id)
comment = Comment.objects.create(
post=post,
content=content,
author_name=author_name,
author_email=author_email
)
return {"id": comment.id}
```
### Form Data
```python
from ninja import Form, Schema, File
from django.core.files.uploadedfile import UploadedFile
class ContactForm(Schema):
name: str
email: str
message: str
@api.post("/contact")
def contact_form(request, data: ContactForm = Form(...)):
"""Handle form submission."""
send_contact_email(data.name, data.email, data.message)
return {"status": "sent"}
# File uploads with form data
@api.post("/upload")
def upload_with_data(
request,
file: UploadedFile = File(...),
description: str = Form(...),
tags: List[str] = Form(default=[])
):
"""Upload file with metadata."""
# Handle file and form data together
pass
```
- ## CRUD Operations
-
- ### Complete CRUD Example
-
- ```python
- from ninja import NinjaAPI, ModelSchema, Schema
- from django.shortcuts import get_object_or_404
- from typing import List, Optional
-
- api = NinjaAPI()
-
- # Schemas
- class CategoryOut(ModelSchema):
- class Config:
- model = Category
- model_fields = ['id', 'name', 'slug']
-
- class PostOut(ModelSchema):
- author: UserSchema
- categories: List[CategoryOut]
-
- class Config:
- model = Post
- model_fields = ['id', 'title', 'slug', 'body', 'status', 'publish']
-
- class PostIn(Schema):
- title: str
- slug: str
- body: str
- status: str = 'draft'
- category_ids: List[int] = []
-
- class PostUpdate(Schema):
- title: Optional[str] = None
- slug: Optional[str] = None
- body: Optional[str] = None
- status: Optional[str] = None
- category_ids: Optional[List[int]] = None
-
- # List
- @api.get("/posts", response=List[PostOut])
- def list_posts(request):
- return Post.objects.select_related('author').prefetch_related('categories').all()
-
- # Retrieve
- @api.get("/posts/{post_id}", response=PostOut)
- def get_post(request, post_id: int):
- post = get_object_or_404(
- Post.objects.select_related('author').prefetch_related('categories'),
- id=post_id
- )
- return post
-
- # Create
- @api.post("/posts", response=PostOut)
- def create_post(request, payload: PostIn):
- post = Post(
- title=payload.title,
- slug=payload.slug,
- body=payload.body,
- status=payload.status,
- author=request.user
- )
- post.save()
-
- if payload.category_ids:
- post.categories.set(payload.category_ids)
-
- return post
-
- # Full Update
- @api.put("/posts/{post_id}", response=PostOut)
- def update_post(request, post_id: int, payload: PostIn):
- post = get_object_or_404(Post, id=post_id)
-
- post.title = payload.title
- post.slug = payload.slug
- post.body = payload.body
- post.status = payload.status
- post.save()
-
- post.categories.set(payload.category_ids)
-
- return post
-
- # Partial Update
- @api.patch("/posts/{post_id}", response=PostOut)
- def partial_update_post(request, post_id: int, payload: PostUpdate):
- post = get_object_or_404(Post, id=post_id)
-
- update_data = payload.dict(exclude_unset=True)
-
- if 'category_ids' in update_data:
- category_ids = update_data.pop('category_ids')
- post.categories.set(category_ids)
-
- for attr, value in update_data.items():
- setattr(post, attr, value)
-
- post.save()
- return post
-
- # Delete
- @api.delete("/posts/{post_id}")
- def delete_post(request, post_id: int):
- post = get_object_or_404(Post, id=post_id)
- post.delete()
- return {"success": True, "id": post_id}
- ```
-
- ### Bulk Operations
-
- ```python
- from ninja import Schema
- from typing import List
-
- class BulkPostCreate(Schema):
- posts: List[PostIn]
-
- @api.post("/posts/bulk", response=List[PostOut])
- def bulk_create_posts(request, payload: BulkPostCreate):
- """Create multiple posts at once."""
- created_posts = []
-
- for post_data in payload.posts:
- post = Post.objects.create(
- **post_data.dict(exclude={'category_ids'}),
- author=request.user
- )
- if post_data.category_ids:
- post.categories.set(post_data.category_ids)
- created_posts.append(post)
-
- return created_posts
-
- @api.post("/posts/bulk-delete")
- def bulk_delete_posts(request, ids: List[int]):
- """Delete multiple posts."""
- deleted, _ = Post.objects.filter(id__in=ids).delete()
- return {"deleted": deleted}
- ```
-
- ## Authentication
-
- ### Global Authentication
-
- ```python
- from ninja import NinjaAPI
- from ninja.security import APIKeyQuery, APIKeyHeader, HttpBearer
-
- # API Key in Header
- class ApiKey(APIKeyHeader):
- param_name = "X-API-Key"
-
- def authenticate(self, request, key):
- try:
- return User.objects.get(api_key=key)
- except User.DoesNotExist:
- pass
-
- # API Key in Query
- class QueryApiKey(APIKeyQuery):
- param_name = "api_key"
-
- def authenticate(self, request, key):
- try:
- return User.objects.get(api_key=key)
- except User.DoesNotExist:
- pass
-
- # Bearer Token (JWT)
- class AuthBearer(HttpBearer):
- def authenticate(self, request, token):
- try:
- payload = jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256'])
- user = User.objects.get(id=payload['user_id'])
- if user.is_active:
- return user
- except (jwt.DecodeError, User.DoesNotExist):
- pass
-
- # Use authentication
- api = NinjaAPI(auth=ApiKey())
- ```
-
- ### Per-Endpoint Authentication
-
- ```python
- from ninja import NinjaAPI
- from ninja.security import django_auth
-
- public_api = NinjaAPI(auth=None) # No auth by default
- private_api = NinjaAPI(auth=AuthBearer())
-
- # Public endpoints
- @public_api.get("/public/data")
- def public_data(request):
- return {"message": "This is public"}
-
- # Private endpoints
- @private_api.get("/private/data", auth=AuthBearer())
- def private_data(request):
- # request.auth contains authenticated user
- return {"user": request.auth.username}
-
- # Mix auth on same API
- api = NinjaAPI()
-
- @api.get("/public")
- def public_endpoint(request):
- return {"public": True}
-
- @api.get("/private", auth=AuthBearer())
- def private_endpoint(request):
- return {"user": request.auth.username}
-
- # Multiple auth methods
- @api.get("/multi-auth", auth=[ApiKey(), AuthBearer()])
- def multi_auth_endpoint(request):
- """Try multiple auth methods."""
- return {"user": request.auth.username}
- ```
-
- ### JWT Authentication
-
- ```python
- from ninja import Schema, NinjaAPI
- from ninja.security import HttpBearer
- import jwt
- from datetime import datetime, timedelta
- from django.conf import settings
-
- class AuthBearer(HttpBearer):
- def authenticate(self, request, token):
- try:
- payload = jwt.decode(
- token,
- settings.SECRET_KEY,
- algorithms=['HS256']
- )
- user = User.objects.get(id=payload['user_id'])
- if user.is_active:
- return user
- except (jwt.ExpiredSignatureError, jwt.InvalidTokenError, User.DoesNotExist):
- return None
-
- class LoginSchema(Schema):
- username: str
- password: str
-
- class TokenSchema(Schema):
- access: str
- refresh: str
- expires_in: int
-
- api = NinjaAPI()
-
- @api.post("/login", response=TokenSchema)
- def login(request, credentials: LoginSchema):
- from django.contrib.auth import authenticate
-
- user = authenticate(
- username=credentials.username,
- password=credentials.password
- )
-
- if not user:
- return {"error": "Invalid credentials"}, 401
-
- # Generate tokens
- access_payload = {
- 'user_id': user.id,
- 'exp': datetime.utcnow() + timedelta(hours=1),
- 'type': 'access'
- }
-
- refresh_payload = {
- 'user_id': user.id,
- 'exp': datetime.utcnow() + timedelta(days=7),
- 'type': 'refresh'
- }
-
- return {
- 'access': jwt.encode(access_payload, settings.SECRET_KEY, algorithm='HS256'),
- 'refresh': jwt.encode(refresh_payload, settings.SECRET_KEY, algorithm='HS256'),
- 'expires_in': 3600
- }
-
- @api.post("/refresh", response=TokenSchema)
- def refresh_token(request, refresh: str = Body(...)):
- try:
- payload = jwt.decode(refresh, settings.SECRET_KEY, algorithms=['HS256'])
-
- if payload.get('type') != 'refresh':
- return {"error": "Invalid token type"}, 401
-
- user = User.objects.get(id=payload['user_id'])
-
- access_payload = {
- 'user_id': user.id,
- 'exp': datetime.utcnow() + timedelta(hours=1),
- 'type': 'access'
- }
-
- return {
- 'access': jwt.encode(access_payload, settings.SECRET_KEY, algorithm='HS256'),
- 'refresh': refresh,
- 'expires_in': 3600
- }
- except jwt.InvalidTokenError:
- return {"error": "Invalid token"}, 401
-
- # Protected endpoints
- @api.get("/profile", auth=AuthBearer())
- def profile(request):
- return {"username": request.auth.username, "email": request.auth.email}
- ```
-
- ### Session Authentication
-
- ```python
- from ninja.security import django_auth
-
- api = NinjaAPI(auth=django_auth)
-
- @api.get("/me")
- def current_user(request):
- # request.user is authenticated via Django session
- return {
- "id": request.user.id,
- "username": request.user.username,
- "email": request.user.email
- }
- ```
-
- ### Basic Authentication
-
- ```python
- from ninja.security import HttpBasicAuth
-
- class BasicAuth(HttpBasicAuth):
- def authenticate(self, request, username, password):
- from django.contrib.auth import authenticate
- user = authenticate(username=username, password=password)
- if user and user.is_active:
- return user
- return None
-
- api = NinjaAPI(auth=BasicAuth())
-
- @api.get("/protected")
- def protected(request):
- return {"user": request.auth.username}
- ```
-
- ### Custom Authentication
-
- ```python
- from ninja.security import APIKeyHeader
- from functools import wraps
-
- class CustomAuth(APIKeyHeader):
- param_name = "Authorization"
-
- def authenticate(self, request, token):
- # Custom token validation
- if not token.startswith("Bearer "):
- return None
-
- token = token[7:] # Remove "Bearer "
-
- try:
- # Custom token validation logic
- user = validate_custom_token(token)
- return user
- except Exception:
- return None
-
- # Function-based auth
- def custom_authenticator(request):
- token = request.headers.get('X-Custom-Token')
- if not token:
- return None
-
- try:
- return validate_token(token)
- except Exception:
- return None
-
- @api.get("/custom-auth", auth=custom_authenticator)
- def custom_auth_endpoint(request):
- return {"authenticated": True}
- ```
-
- ## Pagination
-
- ### LimitOffsetPagination
-
- ```python
- from ninja import NinjaAPI, Schema
- from ninja.pagination import paginate, LimitOffsetPagination
-
- api = NinjaAPI()
-
- class PostOut(ModelSchema):
- class Config:
- model = Post
- model_fields = ['id', 'title', 'slug']
-
- @api.get("/posts", response=List[PostOut])
- @paginate(LimitOffsetPagination)
- def list_posts(request):
- """Returns paginated results with limit and offset."""
- return Post.objects.all()
-
- # Response format:
- # {
- # "items": [...],
- # "count": 100
- # }
- #
- # Query: /posts?limit=10&offset=20
- ```
-
- ### PageNumberPagination
-
- ```python
- from ninja.pagination import PageNumberPagination
-
- class CustomPagination(PageNumberPagination):
- page_size = 20
- page_query_param = 'page'
- page_size_query_param = 'page_size'
- max_page_size = 100
-
- @api.get("/posts", response=List[PostOut])
- @paginate(CustomPagination)
- def list_posts(request):
- """Returns paginated results with page number."""
- return Post.objects.all()
-
- # Response format:
- # {
- # "items": [...],
- # "count": 100,
- # "page": 1,
- # "page_size": 20,
- # "pages": 5
- # }
- #
- # Query: /posts?page=2&page_size=50
- ```
-
- ### Custom Pagination
-
- ```python
- from ninja.pagination import PaginationBase
- from ninja import Schema
- from typing import List, Any
-
- class CustomPaginatedResponse(Schema):
- items: List[Any]
- total: int
- page: int
- per_page: int
- total_pages: int
- has_next: bool
- has_prev: bool
-
- class CursorPagination(PaginationBase):
- """Cursor-based pagination for large datasets."""
-
- class Output(Schema):
- items: List[Any]
- next_cursor: str = None
- prev_cursor: str = None
- has_more: bool
-
- def paginate_queryset(self, queryset, request, **kwargs):
- cursor = request.GET.get('cursor')
- limit = int(request.GET.get('limit', 20))
-
- if cursor:
- queryset = queryset.filter(id__gt=cursor)
-
- items = list(queryset[:limit + 1])
- has_more = len(items) > limit
-
- if has_more:
- items = items[:-1]
-
- return {
- 'items': items,
- 'next_cursor': str(items[-1].id) if items and has_more else None,
- 'prev_cursor': cursor,
- 'has_more': has_more
- }
-
- @api.get("/posts", response=CustomPaginatedResponse)
- @paginate(CursorPagination)
- def list_posts_cursor(request):
- return Post.objects.all().order_by('id')
- ```
-
- ## Error Handling
-
- ### Validation Errors
-
- ```python
- from ninja import NinjaAPI
- from ninja.errors import ValidationError
- from pydantic import ValidationError as PydanticValidationError
-
- api = NinjaAPI()
-
- # Automatic validation
- class UserCreate(Schema):
- username: str # Required
- email: str # Required
- age: int # Must be integer
-
- @api.post("/users")
- def create_user(request, payload: UserCreate):
- # If validation fails, returns 422 with details
- user = User.objects.create_user(**payload.dict())
- return user
-
- # Validation error response format:
- # {
- # "detail": [
- # {
- # "loc": ["body", "username"],
- # "msg": "field required",
- # "type": "value_error.missing"
- # }
- # ]
- # }
- ```
-
- ### Custom Exceptions
-
- ```python
- from ninja import NinjaAPI, HttpError
- from django.core.exceptions import ObjectDoesNotExist
- from django.db import IntegrityError
-
- api = NinjaAPI()
-
- class CustomException(Exception):
- def __init__(self, message, code):
- self.message = message
- self.code = code
-
- @api.exception_handler(CustomException)
- def custom_exception_handler(request, exc):
- return api.create_response(
- request,
- {"error": exc.message, "code": exc.code},
- status=400
- )
-
- @api.exception_handler(ObjectDoesNotExist)
- def not_found_handler(request, exc):
- return api.create_response(
- request,
- {"error": "Resource not found"},
- status=404
- )
-
- @api.exception_handler(IntegrityError)
- def integrity_error_handler(request, exc):
- return api.create_response(
- request,
- {"error": "Database integrity error", "detail": str(exc)},
- status=409
- )
-
- # Using HttpError
- @api.get("/users/{user_id}")
- def get_user(request, user_id: int):
- try:
- user = User.objects.get(id=user_id)
- except User.DoesNotExist:
- raise HttpError(404, "User not found")
- return user
-
- @api.post("/users")
- def create_user(request, payload: UserCreate):
- if User.objects.filter(username=payload.username).exists():
- raise HttpError(409, "Username already exists")
- return User.objects.create_user(**payload.dict())
- ```
-
- ### Error Response Format
-
- ```python
- from ninja import Schema
- from typing import List, Optional
-
- class ErrorDetail(Schema):
- loc: List[str]
- msg: str
- type: str
-
- class ErrorResponse(Schema):
- detail: List[ErrorDetail]
-
- # Configure custom error response
- api = NinjaAPI(
- default_error_responses={
- 400: ErrorResponse,
- 422: ErrorResponse
- }
- )
-
- # Custom error renderer
- def custom_error_renderer(request, errors):
- return {
- "success": False,
- "errors": [
- {
- "field": ".".join(str(loc) for loc in e["loc"]),
- "message": e["msg"]
- }
- for e in errors
- ]
- }
-
- api = NinjaAPI(error_renderer=custom_error_renderer)
- ```
-
- ## File Uploads
-
- ### Single File Upload
-
- ```python
- from ninja import File, NinjaAPI
- from django.core.files.uploadedfile import UploadedFile
-
- api = NinjaAPI()
-
- @api.post("/upload")
- def upload_file(request, file: UploadedFile = File(...)):
- """Upload a single file."""
- # Validate file
- if file.size > 10 * 1024 * 1024: # 10MB
- raise HttpError(413, "File too large (max 10MB)")
-
- allowed_types = ['image/jpeg', 'image/png', 'application/pdf']
- if file.content_type not in allowed_types:
- raise HttpError(415, f"File type {file.content_type} not allowed")
-
- # Save file
- file_path = handle_uploaded_file(file)
-
- return {
- "filename": file.name,
- "size": file.size,
- "content_type": file.content_type,
- "path": file_path
- }
-
- def handle_uploaded_file(f):
- import os
- from django.conf import settings
-
- upload_dir = os.path.join(settings.MEDIA_ROOT, 'uploads')
- os.makedirs(upload_dir, exist_ok=True)
-
- file_path = os.path.join(upload_dir, f.name)
-
- with open(file_path, 'wb+') as destination:
- for chunk in f.chunks():
- destination.write(chunk)
-
- return file_path
- ```
-
- ### Multiple File Uploads
-
- ```python
- from typing import List
-
- @api.post("/upload/multiple")
- def upload_files(request, files: List[UploadedFile] = File(...)):
- """Upload multiple files."""
- results = []
-
- for file in files:
- if file.size > 10 * 1024 * 1024:
- continue # Skip large files
-
- file_path = handle_uploaded_file(file)
- results.append({
- "filename": file.name,
- "size": file.size,
- "path": file_path
- })
-
- return {"uploaded": len(results), "files": results}
- ```
-
- ### File Upload with Schema
-
- ```python
- from ninja import Schema, File, Form
- from typing import Optional
-
- class FileMetadata(Schema):
- title: str
- description: Optional[str] = None
- tags: List[str] = []
-
- @api.post("/upload/with-metadata")
- def upload_with_metadata(
- request,
- file: UploadedFile = File(...),
- metadata: FileMetadata = Form(...)
- ):
- """Upload file with metadata."""
- file_obj = UploadedFile.objects.create(
- file=file,
- title=metadata.title,
- description=metadata.description,
- tags=metadata.tags
- )
- return {"id": file_obj.id, "filename": file.name}
- ```
-
- ## Testing
-
- ### pytest with Django Ninja
-
- ```python
- # tests/test_api.py
- import pytest
- from django.test import Client
- from ninja.testing import TestClient
- from myapp.api import api
-
- @pytest.fixture
- def api_client():
- return TestClient(api)
-
- def test_list_posts(api_client):
- response = api_client.get("/posts")
- assert response.status_code == 200
- assert isinstance(response.json(), list)
-
- def test_create_post(api_client):
- data = {
- "title": "Test Post",
- "slug": "test-post",
- "body": "Test content",
- "status": "draft"
- }
- response = api_client.post("/posts", json=data)
- assert response.status_code == 200
- result = response.json()
- assert result["title"] == "Test Post"
- assert result["id"] is not None
-
- def test_get_post(api_client):
- # First create a post
- create_response = api_client.post("/posts", json={
- "title": "Test",
- "slug": "test",
- "body": "Content"
- })
- post_id = create_response.json()["id"]
-
- # Then retrieve it
- response = api_client.get(f"/posts/{post_id}")
- assert response.status_code == 200
- assert response.json()["title"] == "Test"
-
- def test_update_post(api_client):
- # Create post
- create = api_client.post("/posts", json={
- "title": "Original",
- "slug": "original",
- "body": "Content"
- })
- post_id = create.json()["id"]
-
- # Update
- response = api_client.patch(f"/posts/{post_id}", json={
- "title": "Updated"
- })
- assert response.status_code == 200
- assert response.json()["title"] == "Updated"
-
- def test_delete_post(api_client):
- # Create post
- create = api_client.post("/posts", json={
- "title": "To Delete",
- "slug": "to-delete",
- "body": "Content"
- })
- post_id = create.json()["id"]
-
- # Delete
- response = api_client.delete(f"/posts/{post_id}")
- assert response.status_code == 200
-
- # Verify deleted
- get_response = api_client.get(f"/posts/{post_id}")
- assert get_response.status_code == 404
-
- def test_validation_error(api_client):
- response = api_client.post("/posts", json={
- "title": "Test"
- # Missing required fields
- })
- assert response.status_code == 422
- ```
-
- ### Testing Authentication
-
- ```python
- from ninja.testing import TestClient
- from myapp.api import api, AuthBearer
-
- @pytest.fixture
- def authenticated_client():
- client = TestClient(api)
- # Mock authentication
- user = User.objects.create_user(username="testuser", password="testpass")
- client.force_authenticate(user)
- return client
-
- def test_authenticated_endpoint(authenticated_client):
- response = authenticated_client.get("/profile")
- assert response.status_code == 200
- assert response.json()["username"] == "testuser"
-
- def test_unauthenticated():
- client = TestClient(api)
- response = client.get("/profile")
- assert response.status_code == 401
-
- # Testing with JWT
- def test_jwt_authentication():
- client = TestClient(api)
-
- # Login first
- login_response = client.post("/login", json={
- "username": "testuser",
- "password": "testpass"
- })
- token = login_response.json()["access"]
-
- # Use token
- response = client.get(
- "/profile",
- headers={"Authorization": f"Bearer {token}"}
- )
- assert response.status_code == 200
- ```
-
- ## Async Support
-
- ### Async Views
-
- ```python
- from ninja import NinjaAPI
- from asgiref.sync import sync_to_async
-
- api = NinjaAPI()
-
- @api.get("/async-posts")
- async def async_list_posts(request):
- """Async endpoint for listing posts."""
- # Use async ORM
- posts = await sync_to_async(list)(Post.objects.all()[:10])
- return posts
-
- @api.post("/async-posts")
- async def async_create_post(request, payload: PostCreate):
- """Async endpoint for creating posts."""
- @sync_to_async
- def create_post_sync():
- return Post.objects.create(**payload.dict())
-
- post = await create_post_sync()
- return post
-
- # Async with external API
- import aiohttp
-
- @api.get("/external-data")
- async def fetch_external(request):
- """Fetch data from external API asynchronously."""
- async with aiohttp.ClientSession() as session:
- async with session.get('https://api.example.com/data') as response:
- data = await response.json()
- return data
- ```
-
- ### Async Authentication
-
- ```python
- from ninja.security import HttpBearer
-
- class AsyncAuthBearer(HttpBearer):
- async def authenticate(self, request, token):
- """Async token validation."""
- user = await validate_token_async(token)
- return user
-
- async def validate_token_async(token):
- """Validate JWT token asynchronously."""
- import jwt
- from django.conf import settings
-
- try:
- payload = jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256'])
- user = await sync_to_async(User.objects.get)(id=payload['user_id'])
- return user
- except Exception:
- return None
-
- api = NinjaAPI(auth=AsyncAuthBearer())
-
- @api.get("/async-protected")
- async def async_protected(request):
- return {"user": request.auth.username}
- ```
-
- ## OpenAPI Documentation
-
- ### Swagger UI
-
- ```python
- from ninja import NinjaAPI
-
- api = NinjaAPI(
- title="My API",
- description="API documentation",
- version="1.0.0",
- docs_url="/docs/", # Swagger UI at /api/docs/
- openapi_url="/openapi.json" # OpenAPI schema at /api/openapi.json
- )
-
- # Access at: /api/docs/
- # OpenAPI schema at: /api/openapi.json
- ```
-
- ### Customizing Documentation
-
- ```python
- from ninja import Schema
- from typing import Optional
-
- class PostSchema(Schema):
- id: int
- title: str
- body: str
-
- class Config:
- # OpenAPI schema customization
- schema_extra = {
- "example": {
- "id": 1,
- "title": "My First Post",
- "body": "This is the content of my first post."
- }
- }
-
- @api.get(
- "/posts/{post_id}",
- response=PostSchema,
- summary="Get a post by ID",
- description="Retrieves a specific post by its ID. Returns 404 if not found.",
- tags=["posts"],
- operation_id="get_post_by_id"
- )
- def get_post(request, post_id: int):
- """
- Retrieve a post by ID.
-
- Args:
- post_id: The unique identifier of the post
-
- Returns:
- PostSchema: The requested post
-
- Raises:
- HttpError: 404 if post not found
- """
- post = get_object_or_404(Post, id=post_id)
- return post
-
- # Tags for grouping endpoints
- api = NinjaAPI()
-
- posts_tags = ["posts"]
- users_tags = ["users"]
-
- @api.get("/posts", tags=posts_tags)
- def list_posts(request):
- pass
-
- @api.get("/users", tags=users_tags)
- def list_users(request):
- pass
- ```
-
- ### Response Examples
-
- ```python
- from ninja import Schema
-
- class PostSchema(Schema):
- id: int
- title: str
- body: str
-
- @api.get(
- "/posts/{post_id}",
- response={
- 200: PostSchema,
- 404: ErrorSchema
- },
- examples={
- 200: {"id": 1, "title": "Post Title", "body": "Post content"},
- 404: {"error": "Post not found"}
- }
- )
- def get_post_with_examples(request, post_id: int):
- pass
- ```
-
- ## Django Integration
-
- ### Models
-
- ```python
- # models.py
- from django.db import models
- from django.contrib.auth.models import User
-
- class Post(models.Model):
- STATUS_CHOICES = [
- ('draft', 'Draft'),
- ('published', 'Published'),
- ]
-
- title = models.CharField(max_length=200)
- slug = models.SlugField(unique=True)
- author = models.ForeignKey(User, on_delete=models.CASCADE, related_name='posts')
- body = models.TextField()
- status = models.CharField(max_length=10, choices=STATUS_CHOICES, default='draft')
- categories = models.ManyToManyField('Category', related_name='posts', blank=True)
- created_at = models.DateTimeField(auto_now_add=True)
- updated_at = models.DateTimeField(auto_now=True)
-
- class Meta:
- ordering = ['-created_at']
-
- def __str__(self):
- return self.title
-
- class Category(models.Model):
- name = models.CharField(max_length=100)
- slug = models.SlugField(unique=True)
-
- def __str__(self):
- return self.name
- ```
-
- ### Middleware Integration
-
- ```python
- # middleware.py
- from django.utils.deprecation import MiddlewareMixin
-
- class APIMiddleware(MiddlewareMixin):
- """Custom middleware for API requests."""
-
- def process_request(self, request):
- # Add custom headers or modify request
- request.api_version = request.headers.get('X-API-Version', '1.0')
-
- def process_response(self, request, response):
- # Add headers to all API responses
- if request.path.startswith('/api/'):
- response['X-API-Version'] = getattr(request, 'api_version', '1.0')
- return response
-
- # settings.py
- MIDDLEWARE = [
- # ...
- 'myapp.middleware.APIMiddleware',
- ]
- ```
-
- ### Signals
-
- ```python
- # signals.py
- from django.db.models.signals import post_save
- from django.dispatch import receiver
- from .models import Post
-
- @receiver(post_save, sender=Post)
- def post_created_handler(sender, instance, created, **kwargs):
- """Handle post creation."""
- if created:
- # Send notification, update cache, etc.
- send_notification(instance)
-
- # apps.py
- from django.apps import AppConfig
-
- class MyAppConfig(AppConfig):
- name = 'myapp'
-
- def ready(self):
- import myapp.signals
- ```
-
- ## Best Practices
-
- ### 1. Code Organization
-
- ```python
- # api/__init__.py
- from ninja import NinjaAPI
-
- api = NinjaAPI(title="My API")
-
- # Import routers
- from .users import router as users_router
- from .posts import router as posts_router
-
- api.add_router("/users", users_router)
- api.add_router("/posts", posts_router)
-
- # api/users.py
- from ninja import Router
-
- router = Router()
-
- @router.get("/")
- def list_users(request):
- pass
-
- # api/posts.py
- from ninja import Router
-
- router = Router()
-
- @router.get("/")
- def list_posts(request):
- pass
- ```
-
- ### 2. Schema Organization
-
- ```python
- # schemas/__init__.py
- from .users import UserIn, UserOut, UserUpdate
- from .posts import PostIn, PostOut, PostUpdate
-
- # schemas/users.py
- from ninja import Schema, ModelSchema
- from typing import Optional
-
- class UserBase(Schema):
- username: str
- email: str
- first_name: Optional[str] = None
- last_name: Optional[str] = None
-
- class UserIn(UserBase):
- password: str
-
- class UserOut(UserBase):
- id: int
- created_at: str
-
- class UserUpdate(Schema):
- username: Optional[str] = None
- email: Optional[str] = None
- first_name: Optional[str] = None
- last_name: Optional[str] = None
- ```
-
- ### 3. Error Handling
-
- ```python
- # utils/errors.py
- from ninja import HttpError
-
- class NotFoundError(HttpError):
- def __init__(self, resource, identifier):
- super().__init__(404, f"{resource} with id {identifier} not found")
-
- class PermissionDeniedError(HttpError):
- def __init__(self, message="Permission denied"):
- super().__init__(403, message)
-
- class ValidationError(HttpError):
- def __init__(self, message):
- super().__init__(422, message)
-
- # Usage
- @api.get("/posts/{post_id}")
- def get_post(request, post_id: int):
- post = Post.objects.filter(id=post_id).first()
- if not post:
- raise NotFoundError("Post", post_id)
- if post.author != request.user:
- raise PermissionDeniedError("You can only view your own posts")
- return post
- ```
-
- ### 4. Query Optimization
-
- ```python
- # Always use select_related for ForeignKey
- @api.get("/posts")
- def list_posts(request):
- return Post.objects.select_related('author').all()
-
- # Use prefetch_related for ManyToMany
- @api.get("/posts")
- def list_posts_with_categories(request):
- return Post.objects.select_related('author').prefetch_related('categories').all()
-
- # Use only() for specific fields
- @api.get("/posts")
- def list_posts_minimal(request):
- return Post.objects.only('id', 'title', 'slug').all()
-
- # Use defer() for excluding large fields
- @api.get("/posts")
- def list_posts_without_body(request):
- return Post.objects.defer('body').all()
- ```
-
- ### 5. Caching
-
- ```python
- from django.core.cache import cache
- from functools import wraps
-
- def cache_response(timeout=300):
- """Decorator for caching API responses."""
- def decorator(func):
- @wraps(func)
- def wrapper(request, *args, **kwargs):
- cache_key = f"api:{request.path}:{request.GET.urlencode()}"
- cached = cache.get(cache_key)
- if cached:
- return cached
-
- result = func(request, *args, **kwargs)
- cache.set(cache_key, result, timeout)
- return result
- return wrapper
- return decorator
-
- @api.get("/posts")
- @cache_response(timeout=60)
- def list_posts(request):
- return list(Post.objects.all())
- ```
-
- ## Common Issues
-
- ### Issue: Validation Errors Not Showing
-
- **Problem**: Validation errors return generic 422 without details.
-
- **Solution**:
- ```python
- # Check error renderer configuration
- api = NinjaAPI()
-
- # Default error renderer shows details
- # Custom error renderer for more control
- def custom_error_renderer(errors):
- return {
- "success": False,
- "errors": [
- {"field": e["loc"][-1], "message": e["msg"]}
- for e in errors
- ]
- }
-
- api = NinjaAPI(error_renderer=custom_error_renderer)
- ```
-
- ### Issue: Authentication Not Working
-
- **Problem**: request.auth is None in protected endpoints.
-
- **Solution**:
- ```python
- # Check auth decorator placement
- @api.get("/protected", auth=AuthBearer()) # Correct
- def protected(request):
- return {"user": request.auth.username}
-
- # vs
-
- @api.get("/protected") # Wrong - no auth
- @auth_required # This doesn't work
- def protected(request):
- pass
- ```
-
- ### Issue: File Upload Size Limit
-
- **Problem**: Large file uploads fail silently.
-
- **Solution**:
- ```python
- # settings.py
- DATA_UPLOAD_MAX_MEMORY_SIZE = 10 * 1024 * 1024 # 10MB
- FILE_UPLOAD_MAX_MEMORY_SIZE = 10 * 1024 * 1024 # 10MB
-
- # In the endpoint
- @api.post("/upload")
- def upload_file(request, file: UploadedFile = File(...)):
- if file.size > 10 * 1024 * 1024:
- raise HttpError(413, "File too large")
- ```
-
- ### Issue: CORS Errors
-
- **Problem**: Frontend can't access API due to CORS.
-
- **Solution**:
- ```python
- # Install django-cors-headers
- pip install django-cors-headers
-
- # settings.py
- INSTALLED_APPS = [
- # ...
- 'corsheaders',
- ]
-
- MIDDLEWARE = [
- 'corsheaders.middleware.CorsMiddleware', # Add at the top
- # ...
- ]
-
- CORS_ALLOWED_ORIGINS = [
- "http://localhost:3000",
- "https://myapp.com",
- ]
-
- # Or for development
- CORS_ALLOW_ALL_ORIGINS = True # Only in development!
- ```
-
- ### Issue: Nested Schema Serialization
-
- **Problem**: Nested schemas not serializing correctly.
-
- **Solution**:
- ```python
- # Make sure nested schemas are properly defined
- class AuthorSchema(ModelSchema):
- class Config:
- model = User
- model_fields = ['id', 'username']
+ ## Deep Dives
- class PostSchema(ModelSchema):
- author: AuthorSchema # Must be defined before use
-
- class Config:
- model = Post
- model_fields = ['id', 'title', 'body']
+ For detailed coverage of advanced topics, load these reference files on demand:
- # For many-to-many
- class PostWithCategories(ModelSchema):
- categories: List[str] # Or List[CategorySchema]
-
- class Config:
- model = Post
- model_fields = ['id', 'title']
-
- @staticmethod
- def resolve_categories(obj):
- return [c.name for c in obj.categories.all()]
- ```
+ - **references/auth-pagination-errors.md** — Authentication methods (JWT, API keys, session), pagination strategies (LimitOffset, PageNumber, cursor-based), and error handling patterns
+ - **references/crud-patterns.md** — Complete CRUD operations with Django ORM, bulk operations, and relationship handling
+ - **references/files-async-openapi.md** — File uploads, async view support, and OpenAPI/Swagger documentation customization
+ - **references/django-integration.md** — Django model integration, middleware, signals, and production best practices
+ - **references/testing-troubleshooting.md** — Testing with pytest, authentication testing, and common issue resolutions
## Ecosystem Libraries
### django-ninja-extra
- **URL**: https://github.com/eadwinCode/django-ninja-extra
- **PyPI**: `django-ninja-extra`
- **Version**: 0.31.7 (requires Python >=3.7, Django >=2.2, django-ninja >=1.6.3)
Extension of Django Ninja that adds class-based views (controllers) and advanced features on top of the framework.
**Key features:**
- Class-based controllers via `@api_controller` and HTTP method decorators (`http_get`, etc.), registered with `api.register_controllers()`
- DRF-like permission system: controller-level permissions, route-level overrides, custom `PermissionBase` subclasses
- Dependency injection built on the `Injector` library; injectable service layer (`ModelController`/`ModelService` pattern)
- Inherits Django Ninja's core features (Pydantic validation, async, OpenAPI docs, throttling)
```bash
pip install django-ninja-extra
```
Add `'ninja_extra'` to `INSTALLED_APPS`.
### ninja-schema
- **URL**: https://github.com/eadwinCode/ninja-schema
- **PyPI**: `ninja-schema`
- **Version**: 0.14.3 (requires Python >=3.8)
Converts Django ORM models to Pydantic schemas with full Pydantic feature support. Inspired by django-ninja and djantic.
**Key features:**
- `ModelSchema` with `include`, `exclude`, `optional`, and `depth` config (nested relation schema generation)
- `model_validator` for field-level pre/post validation
- `from_orm()` to instantiate schemas from model instances; `apply_to_model()` to write schema data back to a model instance
- Supports Pydantic v1 and v2 (dual support since 0.13.4)
```bash
pip install ninja-schema
```
### django-ninja-jwt
- **URL**: https://github.com/eadwinCode/django-ninja-jwt
- **PyPI**: `django-ninja-jwt`
- **Version**: 5.4.5 (requires Python >=3.7, `django-ninja-extra>=0.30.5`, `pyjwt>=1.7.1,<3`)
JSON Web Token (JWT) plugin for Django-Ninja. Fork of Jazzband's Simple JWT that removes the DRF dependency and targets Django Ninja. Note: `django-ninja-jwt` depends on `django-ninja-extra`, so installing JWT pulls in Extra.
**Key features:**
- `NinjaJWTDefaultController` registering `obtain_token`, `refresh_token`, and `verify_token` routes
- Custom controllers by inheriting the token controller classes and registering via `api.register_controller()`
- Customizable token classes and claims; configuration via `pydantic-settings`
- Optional use with a plain Django Ninja `Router`
```bash
pip install django-ninja-jwt
```
### django-ninja-aio-crud
- **URL**: https://github.com/caspel26/django-ninja-aio-crud
- **PyPI**: `django-ninja-aio-crud`
- **Version**: 2.34.2 (requires Python >=3.10,<3.15; django-ninja >=1.3.0,<1.7.0)
Async CRUD framework for Django Ninja providing automatic schema generation, filtering, pagination, auth, and M2M management.
**Key features:**
- Fully async CRUD viewsets (create/list/retrieve/update/delete) via `@api.viewset(Model)` and `APIViewSet`
- Two schema styles: meta-driven `Serializer` (existing models) or `ModelSerializer` models with `Read/Create/UpdateSerializer` inner classes
- Auto Pydantic schemas (read/create/update) and dynamic query params via `pydantic.create_model`
- Per-method auth (incl. `AsyncJwtBearer` using `joserfc`), async pagination, M2M relation endpoints with filtering
- Bulk create/update/delete endpoints, `@action`/`@on` custom endpoints, lifecycle hooks
- ORJSON rendering; optional MCP extra exposing ViewSets as Model Context Protocol tools
```bash
pip install django-ninja-aio-crud
```
MCP support: `pip install "django-ninja-aio-crud[mcp]"`
### django-contract-tester
- **URL**: https://github.com/maticardenas/django-contract-tester
- **PyPI**: `django-contract-tester`
- **Version**: 1.8.2
Test utility for validating DRF and Django Ninja test requests/responses against OpenAPI 2.0/3.0.x/3.1.x schemas. Forked from `snok/drf-openapi-tester`.
**Key features:**
- `SchemaTester` with `validate_response()` / `validate_request()`; auto-detects `drf-yasg` or `drf-spectacular` schemas, or loads schema files
- `OpenAPIClient` (extends DRF `APIClient`) and `OpenAPINinjaClient` (extends the Django Ninja test client) for automatic per-request validation
- Built-in key-case testers (`is_camel_case`, `is_pascal_case`, `is_snake_case`, `is_kebab_case`), `ignore_case`, custom validators
- Config file support (`.django-contract-tester` INI or `[tool.django-contract-tester]` in `pyproject.toml`)
```bash
pip install django-contract-tester
```
Optional extras: `django-ninja`, `drf-yasg`, `drf-spectacular`.
+
## References
- **Official Documentation**: https://django-ninja.dev/
- **GitHub Repository**: https://github.com/vitalik/django-ninja
- **Pydantic Documentation**: https://docs.pydantic.dev/
- **OpenAPI Specification**: https://swagger.io/specification/
- **Django Documentation**: https://docs.djangoproject.com/
-