Django REST Framework 企业级 API 架构实战

# 错误的序列化器用法 - 性能杀手
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']

实战：完整 API 实现

用户管理 API

生产级实现需要考虑密码验证、字段限制及动态序列化。

# 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])
    confirm_password = serializers.CharField(write_only=True, required=True)

    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']

    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

# 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
        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
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, AllowAny
from .serializers import UserSerializer
from .permissions import IsOwnerOrReadOnly

class UserViewSet(viewsets.ModelViewSet):
    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()]
        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)
            )
        return queryset.select_related('profile').prefetch_related('groups')

    @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()
        return Response({'message': '注册成功'}, status=status.HTTP_201_CREATED)

分页、过滤、排序

自定义分页器以适配不同的业务场景。

# pagination.py
from rest_framework.pagination import PageNumberPagination
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)
        ]))

# 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')

节流与限流

针对不同 HTTP 方法和用户等级实施差异化限流。

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

class MethodSpecificThrottle(SimpleRateThrottle):
    scope = 'method_specific'
    def get_cache_key(self, request, view):
        ident = request.user.pk if request.user.is_authenticated else self.get_ident(request)
        return self.cache_format % {'scope': self.scope, 'ident': f"{ident}:{request.method}"}
    def get_rate(self):
        if self.request.method == 'GET': return '100/minute'
        elif self.request.method == 'POST': return '20/minute'
        return '10/minute'

# settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'apps.api.throttles.MethodSpecificThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',
        'method_specific': '100/minute',
    }
}

高级实战：企业级 API

缓存优化策略

利用多级缓存和装饰器提升读取性能。

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

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

性能监控中间件

记录请求耗时、数据库查询次数及缓存命中率。

# performance_middleware.py
import time
from django.utils.deprecation import MiddlewareMixin

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
        response['X-Response-Time'] = f'{total_time:.2f}ms'
        response['X-DB-Queries'] = str(db_queries)
        return response

API 版本管理

支持 URL 路径和 Accept Header 两种版本控制策略。

# versioning.py
from rest_framework.versioning import URLPathVersioning, AcceptHeaderVersioning

class NamespaceVersioning(URLPathVersioning):
    default_version = 'v1'
    allowed_versions = ['v1', 'v2', 'v3']
    version_param = 'version'

性能优化指南

数据库优化

避免 N+1 查询是性能优化的关键。

# 优化前 - 产生 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()

序列化器优化

在视图中预加载关联数据，避免在序列化器中执行查询。

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')

缓存策略

实现多级缓存（L1 内存 + L2 Redis）。

# 多级缓存实现
class MultiLevelCache:
    def __init__(self):
        self.l1_cache = caches['memcached']
        self.l2_cache = caches['redis']
    def get(self, key):
        data = self.l1_cache.get(key)
        if data is not None: return data
        data = self.l2_cache.get(key)
        if data is not None:
            self.l1_cache.set(key, data, 60)
            return data
        return None

监控与告警

关键指标监控

使用 Prometheus 收集请求计数、耗时及错误率。

# monitoring.py
from prometheus_client import Counter, Histogram

REQUEST_COUNT = Counter('django_http_requests_total', 'Total HTTP requests', ['method', 'endpoint', 'status'])
REQUEST_DURATION = Histogram('django_http_request_duration_seconds', 'HTTP request duration', ['method', 'endpoint'])

告警配置

定义 Prometheus 规则以触发异常通知。

# prometheus/alerts.yml
groups:
  - name: django_api
    rules:
      - alert: HighErrorRate
        expr: rate(django_http_requests_total{status=~"5.."}[5m]) / rate(django_http_requests_total[5m]) > 0.05
        for: 2m
        labels:
          severity: critical

最佳实践总结

开发规范

1. 代码结构规范：按功能模块划分 apps，如 users, products。
2. API 设计原则：RESTful 风格，资源导向，版本控制从 v1 开始。
3. 安全规范：所有接口默认认证，敏感操作记录日志，输入参数严格验证。

性能优化清单

• 使用 select_related/prefetch_related
• 避免 N+1 查询
• 添加合适索引
• 热点数据缓存
• 异步任务处理

部署配置

生产环境需开启 SSL，配置连接池，并限制渲染器仅使用 JSON。

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

总结

构建高可用 API 需要关注视图集的基础应用、序列化器的性能调优以及细粒度的权限控制。合理分页、强大过滤和频率限制能有效提升用户体验并保障系统安全。持续监控与迭代优化是架构演进的关键。