Django + Ninja API 实战:构建高性能 REST API 的现代指南

By | 2026年10月5日

为什么选择 Django Ninja?

如果你正在使用 Django 开发 REST API,大概率用过 Django REST Framework(DRF)。DRF 功能强大,但代码量较大,性能也有优化空间。Django Ninja 是一个基于类型注解的现代 API 框架,它借鉴了 FastAPI 的许多优秀特性,同时与 Django 生态无缝集成。

Django Ninja 的核心优势:

  • 基于 Pydantic:自动数据验证、序列化和文档生成
  • 类型提示驱动:利用 Python 类型注解,代码简洁且 IDE 友好
  • 高性能:异步支持,性能接近 FastAPI
  • 自动生成 OpenAPI 文档:开箱即用的 Swagger UI 和 ReDoc
  • 与 Django 深度集成:可以直接使用 Django ORM、中间件、认证系统

本文将通过一个完整的「任务管理 API」项目,带你从零掌握 Django Ninja 的实战技巧。

环境准备与项目初始化

首先创建虚拟环境并安装依赖:


python -m venv venv
source venv/bin/activate  # Windows: venv\Scripts\activate
pip install django django-ninja uvicorn

创建 Django 项目和应用:


django-admin startproject taskapi
cd taskapi
python manage.py startapp tasks

在 settings.py 中注册应用:


INSTALLED_APPS = [
    # ...
    'tasks',
]

定义数据模型

编辑 tasks/models.py:


from django.db import models

class Task(models.Model):
    PRIORITY_CHOICES = [
        ('low', 'Low'),
        ('medium', 'Medium'),
        ('high', 'High'),
    ]
    
    title = models.CharField(max_length=200)
    description = models.TextField(blank=True)
    completed = models.BooleanField(default=False)
    priority = models.CharField(max_length=10, choices=PRIORITY_CHOICES, default='medium')
    due_date = models.DateField(null=True, 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

执行迁移:


python manage.py makemigrations
python manage.py migrate

创建 Ninja API 与 Schema

Django Ninja 使用 Pydantic 的 Schema 类来定义请求和响应结构。新建 tasks/schemas.py:


from datetime import date, datetime
from typing import Optional
from ninja import Schema
from pydantic import Field

class TaskIn(Schema):
    title: str = Field(..., max_length=200)
    description: str = ''
    completed: bool = False
    priority: str = 'medium'
    due_date: Optional[date] = None

class TaskOut(Schema):
    id: int
    title: str
    description: str
    completed: bool
    priority: str
    due_date: Optional[date]
    created_at: datetime
    updated_at: datetime

class TaskPatch(Schema):
    title: Optional[str] = None
    description: Optional[str] = None
    completed: Optional[bool] = None
    priority: Optional[str] = None
    due_date: Optional[date] = None

注意:TaskIn 用于创建,TaskPatch 用于部分更新,所有字段可选。TaskOut 用于响应,包含完整字段。

实现 CRUD 接口

创建 tasks/api.py:


from typing import List
from django.shortcuts import get_object_or_404
from ninja import Router
from .models import Task
from .schemas import TaskIn, TaskOut, TaskPatch

router = Router(tags=['tasks'])

@router.post('/', response={201: TaskOut})
def create_task(request, payload: TaskIn):
    task = Task.objects.create(**payload.dict())
    return 201, task

@router.get('/', response=List[TaskOut])
def list_tasks(request, completed: bool = None, priority: str = None):
    qs = Task.objects.all()
    if completed is not None:
        qs = qs.filter(completed=completed)
    if priority:
        qs = qs.filter(priority=priority)
    return qs

@router.get('/{task_id}', response=TaskOut)
def get_task(request, task_id: int):
    return get_object_or_404(Task, id=task_id)

@router.put('/{task_id}', response=TaskOut)
def update_task(request, task_id: int, payload: TaskIn):
    task = get_object_or_404(Task, id=task_id)
    for attr, value in payload.dict().items():
        setattr(task, attr, value)
    task.save()
    return task

@router.patch('/{task_id}', response=TaskOut)
def patch_task(request, task_id: int, payload: TaskPatch):
    task = get_object_or_404(Task, id=task_id)
    for attr, value in payload.dict(exclude_unset=True).items():
        setattr(task, attr, value)
    task.save()
    return task

@router.delete('/{task_id}', response={204: None})
def delete_task(request, task_id: int):
    task = get_object_or_404(Task, id=task_id)
    task.delete()
    return 204, None

关键点:

  • response={201: TaskOut} 指定状态码和响应模型
  • payload.dict(exclude_unset=True) 仅更新传入的字段
  • getobjector_404 自动返回 404 响应

注册 API 路由

在项目根目录的 urls.py 中挂载 Ninja API:


from django.contrib import admin
from django.urls import path
from ninja import NinjaAPI
from tasks.api import router as tasks_router

api = NinjaAPI(title='Task API', version='1.0.0')
api.add_router('/tasks/', tasks_router)

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

启动开发服务器:


python manage.py runserver

访问 http://127.0.0.1:8000/api/docs 即可看到自动生成的 Swagger 文档。

认证与权限控制

Django Ninja 支持多种认证方式。这里以 Token 认证为例:


# tasks/auth.py
from ninja.security import HttpBearer
from django.contrib.auth.models import User

class AuthBearer(HttpBearer):
    def authenticate(self, request, token):
        try:
            return User.objects.get(auth_token=token)
        except User.DoesNotExist:
            return None

在 API 中应用:


from .auth import AuthBearer

@router.post('/', auth=AuthBearer(), response={201: TaskOut})
def create_task(request, payload: TaskIn):
    task = Task.objects.create(owner=request.auth, **payload.dict())
    return 201, task

注意:需要为 User 模型添加 Token 字段,或使用 django-rest-knox 等库。

分页与过滤

对于列表接口,分页是必须的。Django Ninja 提供了内置分页器:


from ninja.pagination import paginate, LimitOffsetPagination

@router.get('/', response=List[TaskOut])
@paginate(LimitOffsetPagination)
def list_tasks(request, completed: bool = None):
    qs = Task.objects.all()
    if completed is not None:
        qs = qs.filter(completed=completed)
    return qs

请求示例:GET /api/tasks/?limit=10&offset=20

响应会包含 items 和 count 字段。

错误处理与自定义异常

统一的错误响应能提升 API 的可用性:


from ninja.errors import HttpError

@router.get('/{task_id}', response=TaskOut)
def get_task(request, task_id: int):
    try:
        return Task.objects.get(id=task_id)
    except Task.DoesNotExist:
        raise HttpError(404, 'Task not found')

全局异常处理:


@api.exception_handler(HttpError)
def http_error_handler(request, exc):
    return api.create_response(request, {'detail': str(exc)}, status=exc.status_code)

性能优化与最佳实践

1. 使用 selectrelated 和 prefetchrelated

如果 Task 有外键关联,列表接口应使用:


qs = Task.objects.select_related('owner').all()

2. 异步视图

Django Ninja 支持异步视图,适合 I/O 密集型操作:


@router.get('/async', response=List[TaskOut])
async def list_tasks_async(request):
    from asgiref.sync import sync_to_async
    tasks = await sync_to_async(list)(Task.objects.all())
    return tasks

3. 使用 ModelSchema 自动生成 Schema


from ninja import ModelSchema
from .models import Task

class TaskOut(ModelSchema):
    class Config:
        model = Task
        model_fields = ['id', 'title', 'description', 'completed', 'priority', 'due_date', 'created_at', 'updated_at']

4. 避免 N+1 查询

在列表接口中,如果序列化时访问关联对象,务必使用 select_related。

5. 使用 django-ninja-extra 增强功能

社区库 django-ninja-extra 提供了类视图、依赖注入等高级特性,适合大型项目。

常见坑与解决方案

  • 坑1:payload.dict() 在 Pydantic v2 中已弃用,应使用 payload.model_dump()。
  • 坑2:日期字段序列化时,确保 due_date 允许 None,否则会报错。
  • 坑3:使用 @paginate 时,返回的 queryset 不能是列表,必须是 QuerySet。
  • 坑4:认证失败时,Ninja 默认返回 401,但不会附带详细信息,可通过自定义异常处理优化。

总结与延伸

本文通过一个完整的任务管理 API,介绍了 Django Ninja 的核心用法:从模型定义、Schema 设计、CRUD 实现,到认证、分页、错误处理和性能优化。Django Ninja 凭借其简洁的语法和强大的功能,非常适合快速构建现代 REST API。

下一步建议:

  • 探索异步 ORM 操作(Django 4.1+ 支持)
  • 集成 django-ninja-jwt 实现 JWT 认证
  • 使用 django-ninja-extra 组织大型项目
  • 编写单元测试,确保 API 稳定性

希望这篇教程能帮助你高效地使用 Django Ninja。如有疑问,欢迎在评论区交流。