DRF 与异步任务

这一课你将能做到

1. 用 Django REST Framework 构建符合 RESTful 规范的 API，覆盖序列化、视图集、路由、权限
2. 用 Celery + Redis 处理耗时任务（发邮件、生成报表、图片处理），不阻塞 HTTP 响应
3. 理解什么时候用 Django 模板渲染（全栈）、什么时候用 DRF（前后端分离）

为什么要学

现代 Web 应用大部分走前后端分离：前端 React/Vue/小程序，后端只暴露 JSON API。Django 原生视图写 API 要手动序列化、手动处理 Content-Type，重复代码多。DRF（Django REST Framework）把这些都封装好，让 Django 变成专业的 API 框架。

和本站（codeforge）的对比：本站用 Spring Boot + Spring Web 写 REST API，用 @RestController + @GetMapping。DRF 用 ViewSet + Router，思路完全一致：声明式路由、自动序列化、权限拦截。Spring 用 Jackson 序列化，DRF 用 Serializer。

异步任务：发邮件、生成报表、图片压缩这类操作可能要几秒到几分钟，不能让 HTTP 请求一直等。Celery 是 Python 生态最成熟的分布式任务队列，把任务丢到 Redis/RabbitMQ，由独立 worker 进程异步执行。

先建立图像

前端 React                DRF                    Celery                Worker
    │                       │                      │                     │
    │  POST /api/articles/  │                      │                     │
    ├──────────────────────>│                      │                     │
    │                       │  serializer.save()   │                     │
    │                       │  ──> 任务落库         │                     │
    │                       │                      │                     │
    │                       │  send_email.delay()  │                     │
    │                       ├─────────────────────>│                     │
    │  201 Created          │                      │ ──> 入队 ──> 取出 ──>│
    │<──────────────────────│                      │                     │
    │  (立即返回，不等邮件)    │                      │                     │ 发邮件

跟着做

步骤 1：安装和配置 DRF

pip install djangorestframework
pip install django-filter  # 可选：强大的过滤

# settings.py
INSTALLED_APPS = [
    ...
    "rest_framework",
    "django_filters",
    "blog",
]

REST_FRAMEWORK = {
    "DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
    "PAGE_SIZE": 20,
    "DEFAULT_FILTER_BACKENDS": [
        "django_filters.rest_framework.DjangoFilterBackend",
        "rest_framework.filters.SearchFilter",
        "rest_framework.filters.OrderingFilter",
    ],
    "DEFAULT_AUTHENTICATION_CLASSES": [
        "rest_framework.authentication.SessionAuthentication",
        "rest_framework.authentication.TokenAuthentication",
    ],
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.IsAuthenticatedOrReadOnly",
    ],
}

步骤 2：序列化器（Serializer）

blog/serializers.py：

from rest_framework import serializers
from .models import Article, Category, Tag
from django.contrib.auth.models import User


class CategorySerializer(serializers.ModelSerializer):
    article_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Category
        fields = ["id", "name", "slug", "article_count"]


class TagSerializer(serializers.ModelSerializer):
    class Meta:
        model = Tag
        fields = ["id", "name"]


class AuthorSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ["id", "username", "first_name"]


class ArticleListSerializer(serializers.ModelSerializer):
    """列表页用轻量序列化器"""
    author = AuthorSerializer(read_only=True)
    category = CategorySerializer(read_only=True)
    tags = TagSerializer(many=True, read_only=True)

    class Meta:
        model = Article
        fields = ["id", "title", "slug", "author", "category", "tags",
                  "views", "status", "created_at"]


class ArticleDetailSerializer(serializers.ModelSerializer):
    """详情页用完整序列化器"""
    author = AuthorSerializer(read_only=True)
    category = CategorySerializer(read_only=True)
    tags = TagSerializer(many=True, read_only=True)
    category_id = serializers.PrimaryKeyRelatedField(
        queryset=Category.objects.all(), source="category", write_only=True
    )
    tag_ids = serializers.PrimaryKeyRelatedField(
        queryset=Tag.objects.all(), source="tags", many=True, write_only=True
    )

    class Meta:
        model = Article
        fields = ["id", "title", "slug", "content", "author", "category",
                  "category_id", "tags", "tag_ids", "views", "status",
                  "created_at", "updated_at"]
        read_only_fields = ["views", "created_at", "updated_at"]

    def validate_slug(self, value):
        qs = Article.objects.filter(slug=value)
        if self.instance:
            qs = qs.exclude(pk=self.instance.pk)
        if qs.exists():
            raise serializers.ValidationError("该 slug 已被使用")
        return value

步骤 3：ViewSet 和 Router

blog/views.py：

from rest_framework import viewsets, filters, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, IsAuthenticatedOrReadOnly
from django_filters.rest_framework import DjangoFilterBackend
from .models import Article, Category
from .serializers import (
    ArticleListSerializer, ArticleDetailSerializer, CategorySerializer
)


class CategoryViewSet(viewsets.ModelViewSet):
    queryset = Category.objects.all()
    serializer_class = CategorySerializer
    lookup_field = "slug"  # 用 slug 而不是 id 作为 URL 标识


class ArticleViewSet(viewsets.ModelViewSet):
    queryset = Article.objects.select_related("author", "category").prefetch_related("tags")
    permission_classes = [IsAuthenticatedOrReadOnly]
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ["status", "category__slug", "author__username"]
    search_fields = ["title", "content"]
    ordering_fields = ["created_at", "views"]
    ordering = ["-created_at"]
    lookup_field = "slug"

    def get_serializer_class(self):
        if self.action == "list":
            return ArticleListSerializer
        return ArticleDetailSerializer

    def perform_create(self, serializer):
        serializer.save(author=self.request.user)

    @action(detail=True, methods=["post"], permission_classes=[IsAuthenticated])
    def like(self, request, slug=None):
        """自定义动作：POST /api/articles/{slug}/like/"""
        article = self.get_object()
        article.views += 1
        article.save(update_fields=["views"])
        return Response({"views": article.views})

    @action(detail=False, methods=["get"])
    def hot(self, request):
        """自定义集合动作：GET /api/articles/hot/"""
        hot = self.get_queryset().filter(views__gt=100)[:10]
        serializer = self.get_serializer(hot, many=True)
        return Response(serializer.data)

路由（用 Router 自动生成）：

# mysite/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from blog.views import ArticleViewSet, CategoryViewSet

router = DefaultRouter()
router.register(r"articles", ArticleViewSet, basename="article")
router.register(r"categories", CategoryViewSet, basename="category")

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/", include(router.urls)),
    path("api-auth/", include("rest_framework.urls")),  # 浏览器登录
]

Router 自动生成的端点：

GET    /api/articles/            → list
POST   /api/articles/            → create
GET    /api/articles/{slug}/     → retrieve
PUT    /api/articles/{slug}/     → update
PATCH  /api/articles/{slug}/     → partial_update
DELETE /api/articles/{slug}/     → destroy
POST   /api/articles/{slug}/like/ → like（自定义）
GET    /api/articles/hot/        → hot（自定义）

步骤 4：Token 认证（适合 SPA / 移动端）

# 在 settings.py 的 INSTALLED_APPS 加
"rest_framework.authtoken",

# 迁移
python manage.py migrate

# 视图：获取 token
from rest_framework.authtoken.views import obtain_auth_token

# urls.py
path("api/token/", obtain_auth_token),

客户端：

# 1. 用账号密码换 token
curl -X POST http://localhost:8000/api/token/ \
     -d "username=alice&password=pass123"
# {"token": "abc123..."}

# 2. 后续请求带 token
curl -H "Authorization: Token abc123..." http://localhost:8000/api/articles/

步骤 5：Celery 异步任务

安装：

pip install celery redis

配置：

# mysite/celery.py
import os
from celery import Celery

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "mysite.settings")

app = Celery("mysite")
app.config_from_object("django.conf:settings", namespace="CELERY")
app.autodiscover_tasks()


# mysite/__init__.py
from .celery import app as celery_app
__all__ = ("celery_app",)

# settings.py
CELERY_BROKER_URL = "redis://localhost:6379/0"
CELERY_RESULT_BACKEND = "redis://localhost:6379/1"
CELERY_ACCEPT_CONTENT = ["json"]
CELERY_TASK_SERIALIZER = "json"
CELERY_RESULT_SERIALIZER = "json"
CELERY_TIMEZONE = "Asia/Shanghai"

定义任务：

# blog/tasks.py
from celery import shared_task
from django.core.mail import send_mail
from .models import Article


@shared_task(bind=True, max_retries=3)
def send_article_notification(self, article_id):
    try:
        article = Article.objects.get(id=article_id)
        send_mail(
            subject=f"新文章发布：{article.title}",
            message=f"作者 {article.author.username} 发布了新文章",
            from_email="noreply@example.com",
            recipient_list=["admin@example.com"],
        )
    except Exception as exc:
        raise self.retry(exc=exc, countdown=60)  # 60 秒后重试


@shared_task
def daily_report():
    from django.db.models import Count
    from django.utils import timezone
    today = timezone.now().date()
    count = Article.objects.filter(created_at__date=today).count()
    # 生成报表、发邮件、写缓存 ...
    return f"今日新增文章：{count}"

视图里调用：

from .tasks import send_article_notification


class ArticleViewSet(viewsets.ModelViewSet):
    def perform_create(self, serializer):
        article = serializer.save(author=self.request.user)
        # 异步发送通知，不阻塞响应
        send_article_notification.delay(article.id)

启动 worker：

# 终端 1：启动 Redis（本地需先装 Redis）
redis-server

# 终端 2：启动 Celery worker
celery -A mysite worker --loglevel=info

# 终端 3（可选）：定时任务调度器
celery -A mysite beat --loglevel=info

定时任务配置：

# settings.py
from celery.schedules import crontab

CELERY_BEAT_SCHEDULE = {
    "daily-report": {
        "task": "blog.tasks.daily_report",
        "schedule": crontab(hour=8, minute=0),  # 每天 8:00
    },
}

讲清楚要点

1. Serializer 是 DRF 的灵魂：类似 Spring 的 DTO + Jackson，负责「模型 → JSON」和「JSON → 模型」双向转换。可以定义多个 serializer 用于不同场景（列表/详情）。
2. ViewSet + Router 消灭 90% 的路由代码：一个 ViewSet 自动产出 CRUD 五个端点。@action 装饰器加自定义动作（如 like、hot）。
3. 权限分两层：permission_classes 控制「能不能访问这个视图」（如 IsAuthenticated）；get_queryset() 控制「能看到哪些数据」（如只看自己的）。两层都要做。
4. perform_create 是写入额外字段的钩子：视图接收的 JSON 通常不含 author，需要在保存时注入 request.user。
5. Celery 解决「HTTP 请求不能慢」的问题：把耗时操作（邮件、报表、第三方 API 调用）丢给后台 worker，HTTP 立即返回。Redis 既当 broker（任务队列）又当 backend（结果存储）。
6. Session 认证 vs Token 认证：Session 适合同站浏览器（带 cookie），Token 适合 SPA/移动端（无状态）。DRF 默认同时支持。

你可能会懵的地方

• DRF 和 Django 原生视图怎么选？ 返回 HTML → Django 原生（TemplateView、ListView）。返回 JSON → DRF。混用时，页面渲染走原生，AJAX 接口走 DRF。

• read_only_fields 和 write_only 的区别？ read_only_fields：序列化时输出，反序列化时忽略（如 created_at）。write_only：反序列化时接受，序列化时不输出（如密码、category_id 这种 ID 字段）。

• Celery 任务失败了怎么办？ 用 bind=True + self.retry() 自动重试。还可以配置 task_annotations 设置不同任务的重试策略。彻底失败的任务进死信队列，需要监控。

• delay() 和 apply_async() 区别？ delay(args) 是 apply_async(args) 的简写。apply_async 支持更多参数：countdown（延迟执行）、eta（指定时间）、expires（过期时间）、queue（指定队列）。

• DRF 分页怎么定制？ DEFAULT_PAGINATION_CLASS 可换成 LimitOffsetPagination（?limit=10&offset=20）或 CursorPagination（高性能、防跳过）。也可以在 ViewSet 上单独覆盖 pagination_class。

错法对照

错法	问题	正解
序列化器里嵌套关系不指定 read_only	POST 时要求传完整嵌套对象	嵌套 serializer 加 read_only=True，写入用 xxx_id 字段
Celery 任务函数传 ORM 对象	序列化失败 / 对象过时	传 ID，任务里重新查询
同步视图里调用耗时 API	用户等几秒，体验差	立即返回，Celery 异步处理
@action 忘记 detail=True/False	路由 404	单对象用 detail=True，集合用 detail=False
queryset 属性不加 select_related	列表接口 N+1	get_queryset() 里加 prefetch
全局 IsAuthenticated 但登录接口也要登录	鸡生蛋问题	登录/注册视图用 AllowAny

卡点急救

1. DRF 浏览器访问 API 返回 403：Session 认证要求 CSRF。浏览器打开 /api-auth/login/ 登录后再访问。
2. Celery 任务不执行：检查 Redis 是否启动，worker 是否在跑，任务是否被 @shared_task 装饰。celery -A mysite inspect active 查看正在执行的任务。
3. AssertionError: 'basename' argument not specified：Router 注册时 queryset 没设置，无法推断 basename。手动加 basename="article"。

面试常问

• DRF 的 Serializer 和 Django Form 有什么区别？ Form 主要处理 HTML 表单，输出带 CSRF 的 HTML。Serializer 处理 JSON/XML 等数据格式，不渲染 HTML，支持嵌套序列化和复杂校验。两者 API 相似（is_valid()、validated_data/cleaned_data），但用途不同。

• Celery 的 broker 和 backend 分别是什么？ broker 是任务队列（Redis/RabbitMQ），存待执行的任务。backend 是结果存储（Redis/数据库），存任务执行结果。可以只配 broker 不配 backend（如果不需要查结果）。

• 如何保证 Celery 任务的幂等性？ 任务可能因重试被多次执行。设计任务时让重复执行结果一致：用唯一约束防重复插入、用 update_or_create 代替 create、用分布式锁保护临界区。

自测三问

1. 一个文章 API 要支持「按分类过滤、按标题搜索、按时间排序」，DRF 需要写哪些配置？
2. 为什么 Celery 任务函数应该传 ID 而不是 ORM 对象？
3. ViewSet 的 get_queryset() 和 permission_classes 在权限控制上分别负责什么？

小结一句话

DRF 让 Django 变成专业的 API 框架（Serializer + ViewSet + Router），Celery 让耗时任务异步执行（Redis + Worker）；两者结合，Django 就能撑起现代前后端分离应用。

延伸

• DRF 官方文档
• Celery 官方文档
• 下一课：测试与部署——pytest、日志、生产环境 checklist。