设计可用的 REST API

学习目标

建立公开列表、公开详情和登录后创建接口，约定 JSON 结构、状态码和分页，使用会话认证时仍保留 CSRF 校验。

为什么接口也需要业务边界

API 是另一种入口，不是绕过表单和权限的捷径。公开接口不能泄露草稿、邮箱或内部权限；创建接口不能接受客户端指定作者。序列化函数明确列出公开字段，PostForm 继续完成业务校验，save_post 继续负责作者和发布事件。

本章用原生 JsonResponse 展示接口原理，不额外引入依赖。DRF 是成熟的第三方选择，适合更复杂的序列化、权限、版本和接口文档，但它不是 Django 核心功能，安装前应核对与当前版本的兼容范围。

具体怎么做

1. 新建 api.py，注册 /api/posts/ 和 /api/posts/<id>/。
2. GET 列表返回 count、page、pages、results；GET 详情只查公开记录。
3. POST 只接受 application/json、字典和白名单字段，限制请求体长度。
4. 认证成功后使用 ModelForm 和业务服务保存，返回 201 与 Location；错误返回明确 error 标识。
5. 新增接口测试，其中一项使用真实 CSRF 检查，先获取带令牌的编辑表单，再发送 X-CSRFToken。

本章新增或替换的完整文件

以下路径均相对于练习项目根目录。替换同名文件的全部内容；未列出的文件保持上一章版本。

blog/api.py

import json
from django.core.paginator import Paginator
from django.http import JsonResponse
from django.shortcuts import get_object_or_404
from django.views.decorators.http import require_http_methods, require_GET
from .models import Post
from .forms import PostForm
from .services import save_post

def serialize(post):
    return {"id": post.pk, "title": post.title, "body": post.body,
            "author": post.author.username, "category": post.category.name,
            "url": post.get_absolute_url(), "created_at": post.created_at.isoformat()}

@require_http_methods(["GET", "POST"])
def posts(request):
    if request.method == "GET":
        queryset = Post.objects.filter(status="published").select_related("author", "category")
        page = Paginator(queryset, 10).get_page(request.GET.get("page"))
        return JsonResponse({"count": page.paginator.count, "page": page.number,
                             "pages": page.paginator.num_pages, "results": [serialize(post) for post in page]})
    if not request.user.is_authenticated:
        return JsonResponse({"error": "authentication_required"}, status=401)
    if request.content_type != "application/json":
        return JsonResponse({"error": "application_json_required"}, status=415)
    if len(request.body) > 64 * 1024:
        return JsonResponse({"error": "body_too_large"}, status=413)
    try:
        data = json.loads(request.body)
    except (ValueError, UnicodeError):
        return JsonResponse({"error": "invalid_json"}, status=400)
    allowed = {"title", "body", "category", "status"}
    if not isinstance(data, dict) or set(data) - allowed or any(not isinstance(value, (str, int)) for value in data.values()):
        return JsonResponse({"error": "invalid_fields"}, status=400)
    form = PostForm(data)
    if not form.is_valid():
        return JsonResponse({"error": "validation_failed", "fields": form.errors.get_json_data()}, status=400)
    post = save_post(form.save(commit=False), request.user)
    response = JsonResponse(serialize(post), status=201)
    response["Location"] = post.get_absolute_url()
    return response

@require_GET
def detail(request, pk):
    try:
        post = Post.objects.select_related("author", "category").get(pk=pk, status="published")
    except Post.DoesNotExist:
        return JsonResponse({"error": "not_found"}, status=404)
    return JsonResponse(serialize(post))

blog/test_api.py

import json
from django.contrib.auth import get_user_model
from django.test import TestCase, Client
from .models import Category, Post

class ApiTests(TestCase):
    def setUp(self):
        self.user = get_user_model().objects.create_user("api-user", "api@example.com")
        self.category = Category.objects.create(name="API", slug="api")
        self.payload = {"title": "通过接口创建文章", "body": "接口正文" * 20, "category": self.category.pk, "status": "published"}

    def test_create_and_public_list(self):
        self.client.force_login(self.user)
        response = self.client.post("/api/posts/", self.payload, content_type="application/json")
        self.assertEqual(response.status_code, 201)
        self.assertEqual(response.json()["author"], self.user.username)
        self.assertEqual(self.client.get("/api/posts/").json()["count"], 1)

    def test_errors_and_draft_do_not_leak(self):
        self.assertEqual(self.client.post("/api/posts/", self.payload, content_type="application/json").status_code, 401)
        self.client.force_login(self.user)
        self.assertEqual(self.client.post("/api/posts/", "bad-json", content_type="application/json").status_code, 400)
        self.assertEqual(self.client.post("/api/posts/", {**self.payload, "author": 99}, content_type="application/json").status_code, 400)
        draft = Post.objects.create(title="草稿", body="私密", author=self.user, category=self.category)
        self.assertEqual(self.client.get(f"/api/posts/{draft.pk}/").status_code, 404)

    def test_session_api_requires_csrf(self):
        client = Client(enforce_csrf_checks=True)
        client.force_login(self.user)
        self.assertEqual(client.post("/api/posts/", self.payload, content_type="application/json").status_code, 403)
        client.get("/posts/new/")
        token = client.cookies["csrftoken"].value
        response = client.post("/api/posts/", self.payload, content_type="application/json", HTTP_X_CSRFTOKEN=token)
        self.assertEqual(response.status_code, 201)

blog/urls.py

from django.urls import path
from . import views, api
app_name = "blog"
urlpatterns = [
    path("api/posts/", api.posts, name="api_posts"),
    path("api/posts/<int:pk>/", api.detail, name="api_detail"),
    path("", views.home, name="home"),
    path("mine/", views.mine, name="mine"),
    path("posts/new/", views.create, name="create"),
    path("posts/<int:pk>/comments/", views.comment, name="comment"),
    path("posts/<int:pk>/edit/", views.edit, name="edit"),
    path("posts/<int:pk>/delete/", views.delete, name="delete"),
    path("posts/<int:pk>/", views.detail, name="detail"),
    path("about/", views.about, name="about"),
    path("old-home/", views.old_home, name="old_home"),
]

如何确认成功

python manage.py test blog.test_api --settings=config.test_settings
curl http://127.0.0.1:8000/api/posts/

列表可匿名读取，只包含已发布文章。登录后打开“写文章”页，在浏览器控制台执行下面完整示例，category 改成后台实际存在的分类 id：

const token = document.querySelector('[name=csrfmiddlewaretoken]').value;
const response = await fetch('/api/posts/', {
  method: 'POST',
  credentials: 'same-origin',
  headers: {'Content-Type': 'application/json', 'X-CSRFToken': token},
  body: JSON.stringify({title: '使用接口发布的学习笔记', body: '这是通过 JSON 接口发布的文章，记录本次练习和验证结果。', category: 1, status: 'published'})
});
console.log(response.status, await response.json());

应返回 201。去掉令牌后返回 403；伪造 author 字段返回 400；读取草稿返回 JSON 404。中间件产生的 CSRF 错误和方法错误仍可能是 HTML，客户端应先检查 Content-Type，不能对所有失败无条件调用 json()。

练习：为“更新自己的文章”设计 PATCH 接口，写出权限、字段白名单、409 冲突与测试计划后再实现。不要把当前公开详情的查询直接复用成任意用户可写的接口。

常见问题

JSON 请求不需要 CSRF 吗？ 使用浏览器会话 Cookie 认证时仍需要。CORS 解决认证吗？ 不解决，它决定浏览器是否允许跨源读取响应。为什么不返回 model.dict？ 这会泄露内部字段并形成不稳定接口契约。

官方参考

• JsonResponse
• CSRF
• Django REST framework 官方站点