用 DRF 搞定企业 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 需要考虑密码强度校验、字段限制和动态权限。下面是一个典型的实现：序列化器里校验两次密码一致，创建用户时使用 create_user；权限上，注册允许任何人，更新和删除只能是用户自己。

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

分页、过滤、排序

默认分页够用，但自定义分页器能让你统一响应格式，比如包装成 {count, next, previous, results}。过滤推荐用 django-filter，既支持精确匹配也支持模糊搜索、范围查询。

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

限流与节流

DRF 的限流可以按用户或 IP 区分。我习惯对不同 HTTP 方法设置不同的频率，比如读接口 100 次/分钟，写接口 20 次/分钟，这样既不影响浏览，又能防止恶意提交。

# 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 的缓存与监控

缓存是提升读取性能最直接的手段。实现多级缓存（本地内存 + Redis）可以在减少网络开销的同时保证数据一致性。针对用户维度的缓存装饰器也很实用，避免重复查询。

# 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 版本更 RESTful，但对调用方要求高一些。通常从 v1 开始，通过命名空间隔离不同版本的视图。

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

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

监控方面，用 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/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

性能优化清单

数据库查询优化是 DRF 性能的基石。始终检查你的 queryset 是否使用了 select_related 和 prefetch_related；谨慎处理序列化器中的关联字段，尽量用注解代替 SerializerMethodField。缓存热点数据，把耗时操作丢进异步队列。生产环境关闭 DEBUG，强制 SSL，限制返回的渲染器只有 JSON。

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

构建高可用 API 没有银弹，都是在对业务理解的基础上，反复权衡一致性和可用性，持续监控和调整。上面这些模式在我维护的项目里帮了不少忙，希望也能让你的 DRF 应用更稳一点。