Django REST Framework 企业级 API 架构实战

错误的序列化器用法往往是性能杀手，例如在 SerializerMethodField 中直接查询数据库导致 N+1 问题：

# 错误的序列化器用法 - 性能杀手
class BadUserSerializer(serializers.ModelSerializer):
    posts = serializers.SerializerMethodField()
    comments = serializers.SerializerMethodField()

    def get_posts(self, obj):
        return obj.posts.count()  # N+1 查询！

    def get_comments(self, obj):
        return obj.comments.count()  # 又一个 N+1！

# 正确的序列化器 - 性能优化
class OptimizedUserSerializer(serializers.ModelSerializer):
    post_count = serializers.IntegerField(source='posts.count', read_only=True)
    comment_count = serializers.IntegerField(source='comments.count', read_only=True)

    class Meta:
        model = User
        fields = ['id', 'name', 'post_count', 'comment_count']
        read_only_fields = ['post_count', 'comment_count']

    def to_representation(self, instance):
        # 预取关联数据
        if not hasattr(instance, '_prefetched_objects_cache'):
            instance = User.objects.prefetch_related('posts', 'comments').get(pk=instance.pk)
        return super().to_representation(instance)

实战：完整 API 实现

用户管理 API

在生产环境中，用户注册和登录需要严格的密码验证和 JWT 支持。

# serializers.py
from rest_framework import serializers
from django.contrib.auth import get_user_model
from django.contrib.auth.password_validation import validate_password

User = get_user_model()

class UserSerializer(serializers.ModelSerializer):
    """用户序列化器 - 生产级实现"""
    password = serializers.CharField(
        write_only=True,
        required=True,
        validators=[validate_password],
        style={'input_type': 'password'}
    )
    confirm_password = serializers.CharField(
        write_only=True,
        required=True,
        style={'input_type': 'password'}
    )

    class Meta:
        model = User
        fields = [
            'id', 'username', 'email', 'password', 'confirm_password',
            'first_name', 'last_name', 'is_active', 'date_joined', 'last_login'
        ]
        read_only_fields = ['id', 'date_joined', 'last_login']
        extra_kwargs = {
            'email': {'required': True},
            'username': {'min_length': 3, 'max_length': 30}
        }

    def validate(self, attrs):
        """验证密码匹配"""
        if attrs['password'] != attrs.get('confirm_password'):
            raise serializers.ValidationError({"password": "两次输入的密码不一致"})
        return attrs

    def create(self, validated_data):
        """创建用户 - 包含密码哈希"""
        validated_data.pop('confirm_password')
        user = User.objects.create_user(**validated_data)
        return user

    def update(self, instance, validated_data):
        """更新用户 - 处理密码更新"""
        validated_data.pop('confirm_password', None)
        password = validated_data.pop('password', None)
        for attr, value in validated_data.items():
            setattr(instance, attr, value)
        if password:
            instance.set_password(password)
            instance.save()
        return instance

权限控制同样关键，我们需要细粒度的访问控制策略：

# 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
        if hasattr(obj, 'user'): return obj.user == request.user
        elif hasattr(obj, 'author'): return obj.author == request.user
        elif hasattr(obj, 'owner'): return obj.owner == request.user
        elif hasattr(obj, 'created_by'): return obj.created_by == request.user
        return False

class RateLimitPermission(permissions.BasePermission):
    """接口调用频率限制"""
    def __init__(self, rate='5/minute'):
        self.rate = rate

    def has_permission(self, request, view):
        cache_key = f"ratelimit:{request.user.id}:{view.__class__.__name__}"
        from django.core.cache import cache
        count = cache.get(cache_key, 0)
        if count >= 5: return False
        cache.set(cache_key, count + 1, 60)
        return True

视图集则负责协调这些组件：

# views.py
from rest_framework import viewsets, status, mixins
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import (
    IsAuthenticated, IsAdminUser, AllowAny
)
from rest_framework_simplejwt.tokens import RefreshToken
from django.utils import timezone
from django.db.models import Q

User = get_user_model()

class UserViewSet(viewsets.ModelViewSet):
    """用户视图集 - 完整的 CRUD 操作"""
    queryset = User.objects.filter(is_active=True)
    serializer_class = UserSerializer

    def get_permissions(self):
        """动态权限控制"""
        if self.action == 'create':
            return [AllowAny()]  # 注册不需要认证
        elif self.action in ['update', 'partial_update', 'destroy']:
            return [IsAuthenticated(), IsOwnerOrReadOnly()]
        else:
            return [IsAuthenticated()]

    def get_queryset(self):
        """查询集优化"""
        queryset = super().get_queryset()
        search = self.request.query_params.get('search')
        if search:
            queryset = queryset.filter(
                Q(username__icontains=search) | Q(email__icontains=search)
            )
        ordering = self.request.query_params.get('ordering', '-date_joined')
        if ordering.lstrip('-') in ['username', 'email', 'date_joined']:
            queryset = queryset.order_by(ordering)
        queryset = queryset.select_related('profile').prefetch_related('groups')
        return queryset

    @action(detail=False, methods=['post'], permission_classes=[AllowAny])
    def register(self, request):
        """用户注册"""
        serializer = self.get_serializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        user = serializer.save()
        refresh = RefreshToken.for_user(user)
        return Response({
            'user': serializer.data,
            'refresh': str(refresh),
            'access': str(refresh.access_token),
        }, status=status.HTTP_201_CREATED)

分页、过滤、排序

自定义分页器可以统一响应格式，提升前端对接效率：

# pagination.py
from rest_framework.pagination import PageNumberPagination
from rest_framework.response import Response
from collections import OrderedDict

class StandardPagination(PageNumberPagination):
    """标准分页器"""
    page_size = 20
    page_size_query_param = 'page_size'
    max_page_size = 100

    def get_paginated_response(self, data):
        return Response(OrderedDict([
            ('count', self.page.paginator.count),
            ('next', self.get_next_link()),
            ('previous', self.get_previous_link()),
            ('results', data)
        ]))

过滤器则利用 django-filter 库实现强大的查询能力：

# filters.py
import django_filters
from django_filters.rest_framework import FilterSet, filters
from django.db.models import Q

class ProductFilter(FilterSet):
    """商品过滤器"""
    name = filters.CharFilter(lookup_expr='icontains')
    min_price = filters.NumberFilter(field_name='price', lookup_expr='gte')
    max_price = filters.NumberFilter(field_name='price', lookup_expr='lte')

    class Meta:
        model = Product
        fields = ['name', 'category', 'status']

    @property
    def qs(self):
        queryset = super().qs
        return queryset.select_related('category').prefetch_related('tags')

节流与限流

防止恶意刷接口是保障服务稳定性的关键，我们可以实现多种限流策略：

# throttles.py
from rest_framework.throttling import SimpleRateThrottle, UserRateThrottle
from django.core.cache import cache

class BurstRateThrottle(UserRateThrottle):
    """突发请求限制"""
    scope = 'burst'
    rate = '100/minute'

class MethodSpecificThrottle(SimpleRateThrottle):
    """按 HTTP 方法限流"""
    scope = 'method_specific'

    def get_rate(self):
        if self.request.method == 'GET': return '100/minute'
        elif self.request.method == 'POST': return '20/minute'
        elif self.request.method == 'DELETE': return '5/minute'
        return '100/minute'

在 settings 中配置默认限流类：

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
        'apps.api.throttles.BurstRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',
        'user': '1000/day',
        'burst': '100/minute',
    }
}

高级实战：企业级 API

缓存优化策略

多级缓存能显著提升读取性能，这里展示一个基于装饰器的缓存方案：

# cache_utils.py
from django.core.cache import cache
from functools import wraps
import hashlib
import json

def cache_per_user(timeout):
    """按用户缓存装饰器"""
    def decorator(view_func):
        @wraps(view_func)
        def _wrapped_view(request, *args, **kwargs):
            cache_key = f"user_cache:{request.user.id}:{request.path}"
            cached_response = cache.get(cache_key)
            if cached_response is not None:
                return cached_response
            response = view_func(request, *args, **kwargs)
            cache.set(cache_key, response, timeout)
            return response
        return _wrapped_view
    return decorator

class CacheMixin:
    """缓存混入类"""
    cache_timeout = 300

    def get_cache_key(self, request):
        key_parts = [self.__class__.__name__, request.method, request.path]
        if request.GET:
            sorted_params = sorted(request.GET.items())
            key_parts.append(hashlib.md5(json.dumps(sorted_params).encode()).hexdigest())
        return ':'.join(key_parts)

性能监控中间件

通过中间件记录每次请求的耗时和数据库查询次数，有助于快速定位瓶颈：

# performance_middleware.py
import time
import logging
from django.utils.deprecation import MiddlewareMixin
from django.db import connection

logger = logging.getLogger('performance')

class PerformanceMiddleware(MiddlewareMixin):
    """性能监控中间件"""
    def process_request(self, request):
        request._start_time = time.time()
        request._db_queries_start = len(connection.queries)

    def process_response(self, request, response):
        total_time = (time.time() - request._start_time) * 1000
        db_queries = len(connection.queries) - request._db_queries_start
        
        log_data = {
            'path': request.path,
            'time_ms': round(total_time, 2),
            'db_queries': db_queries
        }
        
        if total_time > 1000:
            logger.warning(f"慢接口：{json.dumps(log_data)}")
        
        response['X-Response-Time'] = f'{total_time:.2f}ms'
        response['X-DB-Queries'] = str(db_queries)
        return response

API 版本管理

为了向后兼容，API 版本控制必不可少：

# versioning.py
from rest_framework.versioning import URLPathVersioning

class NamespaceVersioning(URLPathVersioning):
    """命名空间版本控制"""
    default_version = 'v1'
    allowed_versions = ['v1', 'v2', 'v3']
    version_param = 'version'

性能优化指南

数据库优化

N+1 查询是 Django 开发中最常见的问题之一，务必使用 select_related 和 prefetch_related：

# 优化前 - 产生 N+1 查询
users = User.objects.all()
for user in users:
    print(user.profile.bio)  # 每次循环都查询数据库

# 优化后 - 使用 select_related
users = User.objects.select_related('profile').all()
for user in users:
    print(user.profile.bio)  # 只查询一次

# 多对多关系使用 prefetch_related
articles = Article.objects.prefetch_related('tags').all()

此外，为常用查询字段添加索引，并使用 values() 仅获取必要字段也能大幅提升性能。

序列化器优化

避免在序列化器内部进行复杂的数据库计算，尽量在视图中完成聚合：

# 优化后 - 高效序列化器
class OptimizedProductSerializer(serializers.ModelSerializer):
    category_name = serializers.CharField(source='category.name', read_only=True)
    review_count = serializers.IntegerField(read_only=True)
    average_rating = serializers.FloatField(read_only=True)

    class Meta:
        model = Product
        fields = ['id', 'name', 'price', 'category_name', 'review_count', 'average_rating']

    @staticmethod
    def setup_eager_loading(queryset):
        return queryset.select_related('category').prefetch_related('reviews')

class ProductViewSet(viewsets.ModelViewSet):
    queryset = Product.objects.all()
    serializer_class = OptimizedProductSerializer

    def get_queryset(self):
        queryset = super().get_queryset()
        queryset = OptimizedProductSerializer.setup_eager_loading(queryset)
        queryset = queryset.annotate(
            review_count=Count('reviews'),
            average_rating=Avg('reviews__rating')
        )
        return queryset

最佳实践总结

开发规范

1. 代码结构规范：按功能模块划分 apps，如 users, products，并在 app 内细分 serializers, views, permissions。
2. API 设计原则：遵循 RESTful 风格，资源导向，错误信息标准化。
3. 安全规范：所有接口默认需要认证，敏感操作记录日志，输入参数严格验证。

部署配置

生产环境配置需特别注意 SSL 重定向、Cookie 安全设置及数据库连接池：

# settings/production.py
DEBUG = False
SECURE_SSL_REDIRECT = True
SESSION_COOKIE_SECURE = True
CSRF_COOKIE_SECURE = True

DATABASES = {
    'default': {
        'CONN_MAX_AGE': 300,
        'OPTIONS': {'sslmode': 'require'}
    }
}

核心要点回顾

• 视图集是基础：合理使用 ModelViewSet 减少重复代码。
• 序列化器是关键：优化序列化性能，避免 N+1 查询。
• 权限是保障：细粒度权限控制保证系统安全。
• 节流是防护：频率限制防止系统被刷爆。

API 设计既是艺术也是科学，不断实践，持续优化，你的架构会越来越优雅。