Django REST Framework 项目教程：从零构建企业级REST API

2026-01-22 04:29:53作者：凤尚柏Louis

还在为Django项目构建API而烦恼？本文将带你全面掌握Django REST Framework，构建强大、可维护的RESTful API服务。

前言：为什么选择Django REST Framework？

Django REST Framework（DRF）是构建Web API的强大工具包，它为Django开发者提供了：

🚀 开箱即用的API功能：序列化、视图、路由、认证等
👀 可浏览的API界面：极大提升开发体验
🔒 完善的安全机制：认证、权限、限流一应俱全
📚 丰富的文档支持：自动生成API文档
🎯 高度可定制性：从简单到复杂场景都能胜任

通过本教程，你将学会：

DRF核心组件的使用方法和最佳实践
如何设计符合RESTful规范的API
实现用户认证和权限控制
处理复杂的数据关系和序列化
构建可扩展的生产级API架构

环境准备与项目初始化

系统要求

Python 3.8+
Django 5.0 或 4.2
Django REST Framework 最新版本

安装与配置

# 创建虚拟环境
python -m venv venv
source venv/bin/activate  # Linux/Mac
# venv\Scripts\activate  # Windows

# 安装依赖
pip install django djangorestframework

# 创建Django项目
django-admin startproject myproject .
cd myproject
django-admin startapp api

基础配置

在 settings.py 中添加DRF配置：

INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    'rest_framework',          # DRF核心
    'api',                     # 我们的API应用
]

REST_FRAMEWORK = {
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticatedOrReadOnly',
    ],
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'rest_framework.authentication.BasicAuthentication',
    ],
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 20
}

核心概念深度解析

1. 序列化器（Serializers）：数据转换的艺术

序列化器是DRF的核心组件，负责：

将复杂数据类型转换为Python原生数据类型
验证传入数据的有效性
处理数据的创建和更新操作

基础序列化器示例

# api/serializers.py
from rest_framework import serializers
from django.contrib.auth.models import User
from .models import Post, Comment

class UserSerializer(serializers.ModelSerializer):
    class Meta:
        model = User
        fields = ['id', 'username', 'email', 'date_joined']

class CommentSerializer(serializers.ModelSerializer):
    author = UserSerializer(read_only=True)
    
    class Meta:
        model = Comment
        fields = ['id', 'content', 'author', 'created_at']

class PostSerializer(serializers.ModelSerializer):
    author = UserSerializer(read_only=True)
    comments = CommentSerializer(many=True, read_only=True)
    comment_count = serializers.SerializerMethodField()
    
    class Meta:
        model = Post
        fields = ['id', 'title', 'content', 'author', 
                 'created_at', 'updated_at', 'comments', 'comment_count']
    
    def get_comment_count(self, obj):
        return obj.comments.count()

序列化器工作流程

flowchart TD
    A[请求数据] --> B[序列化器验证]
    B --> C{验证通过?}
    C -->|是| D[转换为模型实例]
    C -->|否| E[返回错误信息]
    D --> F[保存到数据库]
    F --> G[返回序列化数据]
    E --> H[客户端]
    G --> H

2. 视图（Views）：API的业务逻辑中心

DRF提供了多种视图类型，满足不同复杂度的需求：

视图类型对比表

视图类型	适用场景	优点	缺点
APIView	简单自定义逻辑	完全控制	需要手动实现CRUD
GenericAPIView	基于类的通用视图	代码复用	配置较多
ViewSet	完整CRUD操作	高度封装	灵活性较低
ModelViewSet	模型完整操作	极简代码	定制性有限

视图示例代码

# api/views.py
from rest_framework import viewsets, permissions, status
from rest_framework.decorators import action
from rest_framework.response import Response
from .models import Post, Comment
from .serializers import PostSerializer, CommentSerializer

class PostViewSet(viewsets.ModelViewSet):
    queryset = Post.objects.all()
    serializer_class = PostSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly]
    
    def perform_create(self, serializer):
        serializer.save(author=self.request.user)
    
    @action(detail=True, methods=['post'])
    def add_comment(self, request, pk=None):
        post = self.get_object()
        serializer = CommentSerializer(data=request.data)
        
        if serializer.is_valid():
            serializer.save(author=request.user, post=post)
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

class CommentViewSet(viewsets.ModelViewSet):
    queryset = Comment.objects.all()
    serializer_class = CommentSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly]
    
    def perform_create(self, serializer):
        serializer.save(author=self.request.user)

3. 路由（Routers）：API端点的自动化管理

DRF的路由系统自动为ViewSet生成标准的RESTful端点：

# api/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import PostViewSet, CommentViewSet

router = DefaultRouter()
router.register(r'posts', PostViewSet)
router.register(r'comments', CommentViewSet)

urlpatterns = [
    path('', include(router.urls)),
    path('api-auth/', include('rest_framework.urls', namespace='rest_framework')),
]

自动生成的路由端点

/api/posts/          - GET: 列表, POST: 创建
/api/posts/{id}/     - GET: 详情, PUT: 更新, DELETE: 删除
/api/posts/{id}/add_comment/ - POST: 添加评论（自定义动作）

高级特性实战

1. 认证与权限控制

DRF提供了丰富的认证和权限机制：

# 高级权限配置示例
REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.TokenAuthentication',
        'rest_framework.authentication.SessionAuthentication',
    ],
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',
        'user': '1000/day'
    }
}

自定义权限类

# api/permissions.py
from rest_framework import permissions

class IsOwnerOrReadOnly(permissions.BasePermission):
    """
    自定义权限：只允许对象的所有者进行编辑
    """
    def has_object_permission(self, request, view, obj):
        # 读取权限允许所有请求
        if request.method in permissions.SAFE_METHODS:
            return True
        
        # 写入权限只允许对象的所有者
        return obj.author == request.user

2. 过滤与搜索

实现强大的数据过滤功能：

# 安装过滤依赖
pip install django-filter

# settings.py配置
INSTALLED_APPS = [
    ...,
    'django_filters',
]

REST_FRAMEWORK = {
    'DEFAULT_FILTER_BACKENDS': [
        'django_filters.rest_framework.DjangoFilterBackend',
        'rest_framework.filters.SearchFilter',
        'rest_framework.filters.OrderingFilter',
    ]
}

# 视图中的使用
class PostViewSet(viewsets.ModelViewSet):
    # ...
    filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
    filterset_fields = ['author', 'created_at']
    search_fields = ['title', 'content']
    ordering_fields = ['created_at', 'updated_at']
    ordering = ['-created_at']

3. 分页处理

处理大数据集的分页显示：

# 自定义分页类
from rest_framework.pagination import PageNumberPagination

class LargeResultsSetPagination(PageNumberPagination):
    page_size = 100
    page_size_query_param = 'page_size'
    max_page_size = 1000

class StandardResultsSetPagination(PageNumberPagination):
    page_size = 20
    page_size_query_param = 'page_size'
    max_page_size = 100

# 在视图中使用
class PostViewSet(viewsets.ModelViewSet):
    pagination_class = StandardResultsSetPagination
    # ...

项目实战：博客API系统

让我们构建一个完整的博客API系统，包含以下功能：

用户认证和管理
文章CRUD操作
评论系统
标签分类
搜索和过滤

数据模型设计

# api/models.py
from django.db import models
from django.contrib.auth.models import User

class Tag(models.Model):
    name = models.CharField(max_length=50, unique=True)
    created_at = models.DateTimeField(auto_now_add=True)

    def __str__(self):
        return self.name

class Post(models.Model):
    title = models.CharField(max_length=200)
    content = models.TextField()
    author = models.ForeignKey(User, on_delete=models.CASCADE)
    tags = models.ManyToManyField(Tag, blank=True)
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    is_published = models.BooleanField(default=False)
    
    class Meta:
        ordering = ['-created_at']
    
    def __str__(self):
        return self.title

class Comment(models.Model):
    post = models.ForeignKey(Post, on_delete=models.CASCADE, related_name='comments')
    author = models.ForeignKey(User, on_delete=models.CASCADE)
    content = models.TextField()
    created_at = models.DateTimeField(auto_now_add=True)
    updated_at = models.DateTimeField(auto_now=True)
    
    class Meta:
        ordering = ['created_at']
    
    def __str__(self):
        return f"Comment by {self.author} on {self.post}"

完整的序列化器配置

# api/serializers.py
from rest_framework import serializers
from django.contrib.auth.models import User
from .models import Post, Comment, Tag

class UserSerializer(serializers.ModelSerializer):
    post_count = serializers.SerializerMethodField()
    
    class Meta:
        model = User
        fields = ['id', 'username', 'email', 'date_joined', 'post_count']
        read_only_fields = ['date_joined']
    
    def get_post_count(self, obj):
        return obj.post_set.count()

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

class CommentSerializer(serializers.ModelSerializer):
    author = UserSerializer(read_only=True)
    
    class Meta:
        model = Comment
        fields = ['id', 'content', 'author', 'created_at', 'updated_at']
        read_only_fields = ['author', 'created_at', 'updated_at']

class PostSerializer(serializers.ModelSerializer):
    author = UserSerializer(read_only=True)
    comments = CommentSerializer(many=True, read_only=True)
    tags = TagSerializer(many=True, read_only=True)
    tag_ids = serializers.PrimaryKeyRelatedField(
        many=True, 
        queryset=Tag.objects.all(), 
        source='tags',
        write_only=True
    )
    comment_count = serializers.SerializerMethodField()
    read_time = serializers.SerializerMethodField()
    
    class Meta:
        model = Post
        fields = [
            'id', 'title', 'content', 'author', 'tags', 'tag_ids',
            'created_at', 'updated_at', 'is_published',
            'comments', 'comment_count', 'read_time'
        ]
        read_only_fields = ['author', 'created_at', 'updated_at']
    
    def get_comment_count(self, obj):
        return obj.comments.count()
    
    def get_read_time(self, obj):
        # 假设阅读速度为200字/分钟
        word_count = len(obj.content.split())
        return max(1, round(word_count / 200))
    
    def create(self, validated_data):
        tags = validated_data.pop('tags', [])
        post = Post.objects.create(**validated_data)
        post.tags.set(tags)
        return post
    
    def update(self, instance, validated_data):
        tags = validated_data.pop('tags', None)
        instance = super().update(instance, validated_data)
        if tags is not None:
            instance.tags.set(tags)
        return instance

完整的视图配置

# api/views.py
from rest_framework import viewsets, permissions, status, filters
from rest_framework.decorators import action
from rest_framework.response import Response
from django_filters.rest_framework import DjangoFilterBackend
from django.contrib.auth.models import User
from .models import Post, Comment, Tag
from .serializers import (
    PostSerializer, CommentSerializer, 
    TagSerializer, UserSerializer
)
from .permissions import IsOwnerOrReadOnly
from .pagination import StandardResultsSetPagination

class UserViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = User.objects.all()
    serializer_class = UserSerializer
    pagination_class = StandardResultsSetPagination

class TagViewSet(viewsets.ModelViewSet):
    queryset = Tag.objects.all()
    serializer_class = TagSerializer
    permission_classes = [permissions.IsAdminUser]
    pagination_class = StandardResultsSetPagination

class PostViewSet(viewsets.ModelViewSet):
    queryset = Post.objects.all()
    serializer_class = PostSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly, IsOwnerOrReadOnly]
    pagination_class = StandardResultsSetPagination
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_fields = ['author', 'tags', 'is_published', 'created_at']
    search_fields = ['title', 'content']
    ordering_fields = ['created_at', 'updated_at', 'title']
    ordering = ['-created_at']
    
    def perform_create(self, serializer):
        serializer.save(author=self.request.user)
    
    @action(detail=True, methods=['post'])
    def publish(self, request, pk=None):
        post = self.get_object()
        if post.author != request.user and not request.user.is_staff:
            return Response(
                {'error': '只有作者或管理员可以发布文章'},
                status=status.HTTP_403_FORBIDDEN
            )
        post.is_published = True
        post.save()
        return Response({'status': '文章已发布'})
    
    @action(detail=True, methods=['post'])
    def unpublish(self, request, pk=None):
        post = self.get_object()
        if post.author != request.user and not request.user.is_staff:
            return Response(
                {'error': '只有作者或管理员可以取消发布'},
                status=status.HTTP_403_FORBIDDEN
            )
        post.is_published = False
        post.save()
        return Response({'status': '文章已取消发布'})
    
    @action(detail=False)
    def published(self, request):
        published_posts = self.get_queryset().filter(is_published=True)
        page = self.paginate_queryset(published_posts)
        serializer = self.get_serializer(page, many=True)
        return self.get_paginated_response(serializer.data)

class CommentViewSet(viewsets.ModelViewSet):
    queryset = Comment.objects.all()
    serializer_class = CommentSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly, IsOwnerOrReadOnly]
    pagination_class = StandardResultsSetPagination
    
    def perform_create(self, serializer):
        serializer.save(author=self.request.user)
    
    @action(detail=True, methods=['post'])
    def like(self, request, pk=None):
        comment = self.get_object()
        # 实现点赞逻辑
        return Response({'status': '点赞成功'})

路由配置

# api/urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import PostViewSet, CommentViewSet, TagViewSet, UserViewSet

router = DefaultRouter()
router.register(r'users', UserViewSet)
router.register(r'posts', PostViewSet)
router.register(r'comments', CommentViewSet)
router.register(r'tags', TagViewSet)

urlpatterns = [
    path('', include(router.urls)),
    path('api-auth/', include('rest_framework.urls', namespace='rest_framework')),
]

主项目URL配置

# myproject/urls.py
from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path('admin/', admin.site.urls),
    path('api/', include('api.urls')),
    path('api-auth/', include('rest_framework.urls')),
]

测试与部署

编写单元测试

# api/tests.py
from django.urls import reverse
from rest_framework.test import APITestCase
from rest_framework import status
from django.contrib.auth.models import User
from .models import Post, Tag

class PostAPITestCase(APITestCase):
    def setUp(self):
        self.user = User.objects.create_user(
            username='testuser', 
            password='testpass123'
        )
        self.tag = Tag.objects.create(name='Django')
        self.post = Post.objects.create(
            title='Test Post',
            content='Test content',
            author=self.user
        )
        self.post.tags.add(self.tag)
    
    def test_get_posts(self):
        url = reverse('post-list')
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
    
    def test_create_post_authenticated(self):
        self.client.force_authenticate(user=self.user)
        url = reverse('post-list')
        data = {
            'title': 'New Post',
            'content': 'New content',
            'tag_ids': [self.tag.id]
        }
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
    
    def test_create_post_unauthenticated(self):
        url = reverse('post-list')
        data = {'title': 'New Post', 'content': 'New content'}
        response = self.client.post(url, data)
        self.assertEqual(response.status_code, status.HTTP_403_FORBIDDEN)

部署配置

创建生产环境配置：

# settings/production.py
from .base import *

DEBUG = False
ALLOWED_HOSTS = ['yourdomain.com', 'www.yourdomain.com']

# 数据库配置
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'NAME': os.getenv('DB_NAME'),
        'USER': os.getenv('DB_USER'),
        'PASSWORD': os.getenv('DB_PASSWORD'),
        'HOST': os.getenv('DB_HOST'),
        'PORT': os.getenv('DB_PORT'),
    }
}

# 静态文件配置
STATIC_ROOT = os.path.join(BASE_DIR, 'staticfiles')
STATIC_URL = '/static/'

# 安全配置
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True
SECURE_BROWSER_XSS_FILTER = True
SECURE_CONTENT_TYPE_NOSNIFF = True

性能优化与最佳实践

1. 查询优化

# 使用select_related和prefetch_related优化查询
class PostViewSet(viewsets.ModelViewSet):
    def get_queryset(self):
        return Post.objects.select_related('author').prefetch_related(
            'tags', 'comments', 'comments__author'
        )

2. 缓存策略

# 使用DRF的缓存扩展
from rest_framework_extensions.cache.mixins import CacheResponseMixin

class CachedPostViewSet(CacheResponseMixin, PostViewSet):
    cache_timeout = 60 * 15  # 15分钟缓存

3. 异步处理

# 使用Django的异步支持
from asgiref.sync import async_to_sync

class PostViewSet(viewsets.ModelViewSet):
    @action(detail=True, methods=['post'])
    @async_to_sync
    async def async_action(self, request, pk=None):
        # 异步处理逻辑
        pass