【Django REF】Django REF 常用知识点汇总

1. 序列化器（Serializers）

1.1 自定义字段

1.1.1、直接继承**serializers.Field**并重写关键方法

通过继承serializers.Field类，并重写to_representation和to_internal_value方法来实现自定义序列化逻辑。to_representation用于控制从Python对象到原始数据类型的转换（例如，在返回给客户端之前将数据库中的datetime对象格式化为字符串），而to_internal_value用于控制从原始数据类型到Python对象的转换（例如，将输入字符串解析为datetime对象）。

from rest_framework import serializers
from datetime import datetime
class CustomDateField(serializers.Field):
    """
    自定义日期字段处理类
    功能：
    - 将数据库中的datetime对象格式化为字符串输出
    - 将输入字符串解析为datetime对象
    """
    def to_representation(self, value):
        """将存储的datetime对象转换为字符串"""
        # 使用strftime进行格式化，确保格式统一
        return value.strftime("%Y-%m-%d %H:%M:%S") if value else None
    def to_internal_value(self, data):
        """将输入字符串转换为datetime对象"""
        try:
            # 严格校验日期格式，防止非法输入
            return datetime.strptime(data, "%Y-%m-%d %H:%M:%S")
        except (ValueError, TypeError) as e:
            # 抛出验证错误，会被序列化器的is_valid()捕获
            raise serializers.ValidationError(f"日期格式错误，需要'YYYY-MM-DD HH:MM:SS'格式。错误：{str(e)}")

1.1.2、使用已有的字段类型并扩展其功能

有时可能想要基于现有的字段类型（如serializers.CharField, serializers.IntegerField等）进行扩展。你可以继承这些具体的字段类，并根据需要添加或修改功能。

rom rest_framework import serializers

class PositiveIntegerField(serializers.IntegerField):
    def to_internal_value(self, data):
        value = super().to_internal_value(data)
        if value < 0:
            raise serializers.ValidationError("This field must be a positive integer.")
        return value
class ExampleSerializer(serializers.Serializer):
    id = PositiveIntegerField()

• 定义了一个名为 PositiveIntegerField 的类，继承自 IntegerField。
• IntegerField 是 Django REST Framework 提供的标准字段，用于处理整数类型的验证。
• to_internal_value 是 DRF 中用于将输入数据（如前端传入的值）转换为内部表示的方法。
• 在序列化器中，to_internal_value 方法用于对输入数据进行验证和转换。

value = super().to_internal_value(data)

• 调用 IntegerField 的 to_internal_value 方法，先对输入数据进行整数类型的验证。
• 如果输入数据不是整数，会在此处抛出验证错误。

if value < 0:
    raise serializers.ValidationError("This field must be a positive integer.")

• 验证输入的整数值是否小于 0。
• 如果值小于 0，抛出一个 ValidationError，并附带错误提示信息 "This field must be a positive integer."。

1.1.3、利用SerializerMethodField创建只读字段

当需要对序列化过程中的某个字段应用复杂的逻辑时，可以使用serializers.SerializerMethodField。这允许你在序列化器中定义一个方法，该方法将被用来生成字段值。

from rest_framework import serializers

class UserSerializer(serializers.Serializer):
    username = serializers.CharField()
    email = serializers.EmailField()
    full_name = serializers.SerializerMethodField()  # 只读字段

    def get_full_name(self, obj):
        """
        自定义方法，根据对象生成全名。
        """
        return f"{obj.first_name} {obj.last_name}"

请注意，在上面的例子中，get_full_name方法会接收序列化过程中的对象实例作为参数，并返回期望的值。这个方法的名字必须遵循get_<field_name>的命名规则。

1.1.4、使用**serializers.Serializer**作为字段

from rest_framework import serializers

class AddressSerializer(serializers.Serializer):
    street = serializers.CharField()
    city = serializers.CharField()
    state = serializers.CharField()

class UserSerializer(serializers.Serializer):
    name = serializers.CharField()
    email = serializers.EmailField()
    address = AddressSerializer()  # 使用另一个Serializer作为字段

1.1.5、格式化字段（datetime）

class LoginlogSerializer(serializers.ModelSerializer):
    login_time=serializers.DateTimeField(format='%Y年%m月%d日 %H:%M:%S',read_only=True)

    class Meta:
        model=XCLoginlog
        fields=('id','device_type','ip_address','login_time','os','post_data','success','user','user_agent')

这行代码使用 Django REST Framework 的 DateTimeField 来定义一个序列化字段。format 参数指定了将 datetime 对象序列化为字符串时的格式，read_only=True 表示该字段仅用于序列化输出，不用于反序列化输入。

1.2 嵌套序列化器

from rest_framework import serializers
from .models import User, Profile

class ProfileSerializer(serializers.ModelSerializer):
    """用户资料序列化器，用于将Profile模型实例与Python原生数据类型之间进行转换"""
    class Meta:
        # 指定关联的模型为Profile
        model = Profile
        # 定义序列化和反序列化时包含的字段
        fields = ["bio", "website"]
        # 扩展配置：设置字段额外属性
        extra_kwargs = {
            # 允许bio字段为空字符串，且该字段不是必需的
            'bio': {'allow_blank': True, 'required': False},
            # 为website字段添加URL验证器，确保输入的是有效的URL
            'website': {'validators': [serializers.URLValidator()]}
        }

class UserSerializer(serializers.ModelSerializer):
    """用户主序列化器，用于将User模型实例与Python原生数据类型之间进行转换"""
    # 将ProfileSerializer作为嵌套序列化器，设置为非必须字段以支持部分更新
    profile = ProfileSerializer(required=False)

    class Meta:
        # 指定关联的模型为User
        model = User
        # 定义序列化和反序列化时包含的字段
        fields = ["username", "email", "profile"]
        # 扩展配置：设置字段验证规则
        extra_kwargs = {
            'email': {
                # 邮箱字段是必需的
                'required': True,
                # 为邮箱字段添加唯一验证器，确保该邮箱在所有用户中是唯一的
                'validators': [
                    serializers.UniqueValidator(
                        queryset=User.objects.all(),
                        message="该邮箱已被注册"
                    )
                ]
            }
        }

    def create(self, validated_data):
        """
        重写创建方法以处理嵌套关系
        当使用该序列化器创建新的User实例时，此方法会被调用
        """
        # 从验证后的数据中弹出profile数据，如果不存在则返回None
        profile_data = validated_data.pop('profile', None)
        # 使用剩余的验证数据创建User实例
        user = User.objects.create(**validated_data)
        # 如果存在profile数据
        if profile_data:
            # 创建与该用户关联的Profile实例
            Profile.objects.create(user=user, **profile_data)
        # 返回创建好的User实例
        return user

    def update(self, instance, validated_data):
        """
        重写更新方法以处理嵌套关系
        当使用该序列化器更新已有的User实例时，此方法会被调用
        """
        # 从验证后的数据中弹出profile数据，如果不存在则返回空字典
        profile_data = validated_data.pop('profile', {})
        # 调用父类的update方法更新用户实例
        instance = super().update(instance, validated_data)
        # 获取用户关联的Profile实例
        profile = instance.profile
        # 如果用户已经有关联的Profile实例
        if profile:
            # 遍历profile数据中的每个属性和值
            for attr, value in profile_data.items():
                # 为Profile实例设置新的属性值
                setattr(profile, attr, value)
            # 保存更新后的Profile实例
            profile.save()
        else:
            # 如果用户没有关联的Profile实例，则创建一个新的Profile实例并关联到该用户
            Profile.objects.create(user=instance, **profile_data)
        # 返回更新后的User实例
        return instance

自己编写代码的案例，权限管理系统中菜单序列化

from rest_framework import serializers
from userapi.models import XCMenu,XCRole


class MenuSerializer(serializers.ModelSerializer):
    create_time=serializers.DateTimeField(format='%Y年%m月%d日 %H:%M:%S',read_only=True)
    update_time=serializers.DateTimeField(format='%Y年%m月%d日 %H:%M:%S',read_only=True)
    creator = serializers.SerializerMethodField()
    children = serializers.SerializerMethodField()
    is_top_level = serializers.SerializerMethodField()


    class Meta:
        model = XCMenu
        fields = ['id','name','name_zh','create_time','update_time','creator','path','redrect','component','meta','status',"is_top_level",'parent','children']
    def get_creator(self, obj):
        # 返回 creator 的 username 字段
        return obj.creator.username if obj.creator else None
    def get_children(self, obj):
        # 递归获取子菜单
        # 获取当前菜单的子菜单
        children = XCMenu.objects.filter(parent=obj)
        # 使用 MenuSerializer 序列化子菜单
        return MenuSerializer(children, many=True).data

    def get_is_top_level(self, obj):
        # 如果菜单没有父菜单，则为顶级菜单
        return obj.parent is None

1.3 动态字段

class DynamicFieldsModelSerializer(serializers.ModelSerializer):
    """
    支持动态字段控制的序列化器基类
    功能：
    - 允许通过fields参数指定需要的字段
    - 允许通过exclude参数指定排除的字段
    """
    def init(self, *args, **kwargs):
        # 从参数中提取字段控制参数
        fields = kwargs.pop('fields', None)
        exclude = kwargs.pop('exclude', None)
        super().__init__(*args, **kwargs)
        
        if fields is not None and exclude is not None:
            raise serializers.ValidationError("不能同时指定fields和exclude参数")
        
            if fields is not None:
                # 白名单模式：只保留指定字段
                allowed = set(fields)
                existing = set(self.fields)
                for field_name in existing - allowed:
                    self.fields.pop(field_name)
    
        if exclude is not None:
            # 黑名单模式：排除指定字段
            excluded = set(exclude)
            for field_name in excluded:
                self.fields.pop(field_name, None)  # 安全删除字段

2. 视图（Views）

2.1、视图集（ViewSets）

from rest_framework import viewsets, permissions
from .models import User
from .serializers import UserSerializer
class UserViewSet(viewsets.ModelViewSet):
    """
    用户视图集（完整CRUD）
    功能：
    - 自动生成标准REST接口
    - 包含列表、详情、创建、更新、删除操作
    """
    queryset = User.objects.all().select_related('profile')  # 优化查询性能
    serializer_class = UserSerializer
    permission_classes = [permissions.IsAuthenticatedOrReadOnly]  # 认证用户可写，匿名用户只读
    filter_backends = [filters.OrderingFilter]  # 启用排序
    ordering_fields = ['date_joined', 'username']  # 允许排序的字段
    
    def get_queryset(self):
        """重写查询集以实现业务逻辑过滤"""
        queryset = super().get_queryset()
        # 示例：根据请求参数过滤活跃用户
        is_active = self.request.query_params.get('active', None)
        if is_active in ['true', 'false']:
            queryset = queryset.filter(is_active=is_active.lower() == 'true')
        return queryset

重写视图基

1. **list** 方法：用于获取所有书籍的列表。重写该方法时，我们可以添加额外的逻辑，如过滤、排序等，并对返回的数据进行额外处理。
2. **create** 方法：用于创建新的书籍。重写该方法时，我们可以添加额外的逻辑，如记录日志、发送通知等。
3. **retrieve** 方法：用于获取单个书籍的详细信息。重写该方法时，我们可以添加额外的逻辑，如检查权限、增加访问计数等。
4. **update** 方法：用于更新书籍的信息。重写该方法时，我们可以添加额外的逻辑，如记录更新日志等。
5. **destroy** 方法：用于删除书籍。重写该方法时，我们可以添加额外的逻辑，如检查是否可以删除、记录删除日志等。
6. **recent_books** 方法：这是一个自定义动作，用于获取最近出版的书籍。使用 @action 装饰器可以将自定义动作添加到视图集中。

from rest_framework import viewsets
from .models import Book
from .serializers import BookSerializer
from rest_framework.response import Response
from rest_framework import status

class BookViewSet(viewsets.ModelViewSet):
    # 指定视图集要处理的查询集，这里获取所有的 Book 模型实例
    queryset = Book.objects.all()
    # 指定用于序列化和反序列化的序列化器类
    serializer_class = BookSerializer

    # 重写 list 方法，该方法用于处理获取所有书籍列表的请求
    def list(self, request, *args, **kwargs):
        # 可以在这里添加额外的逻辑，例如过滤、排序等
        # 调用父类的过滤方法对查询集进行过滤
        queryset = self.filter_queryset(self.get_queryset())
        # 对过滤后的查询集进行分页处理
        page = self.paginate_queryset(queryset)
        # 如果分页结果不为空
        if page is not None:
            # 使用序列化器对分页后的结果进行序列化，many=True 表示处理多个对象
            serializer = self.get_serializer(page, many=True)
            # 返回分页后的响应数据
            return self.get_paginated_response(serializer.data)
        # 如果没有进行分页，直接对查询集进行序列化
        serializer = self.get_serializer(queryset, many=True)
        # 可以在这里对返回的数据进行额外处理
        # 自定义响应数据的结构，包含状态信息和实际数据
        response_data = {
            'status': 'success',
            'data': serializer.data
        }
        return Response(response_data)

    # 重写 create 方法，该方法用于处理创建新书籍的请求
    def create(self, request, *args, **kwargs):
        # 使用请求数据创建序列化器实例
        serializer = self.get_serializer(data=request.data)
        # 验证请求数据的有效性，如果无效则抛出异常
        serializer.is_valid(raise_exception=True)
        # 可以在这里添加额外的逻辑，例如记录日志、发送通知等
        # 调用父类的创建方法创建新的书籍实例
        self.perform_create(serializer)
        # 获取成功响应的头部信息
        headers = self.get_success_headers(serializer.data)
        # 自定义响应数据的结构，包含状态信息、消息和实际数据
        response_data = {
            'status': 'success',
            'message': 'Book created successfully',
            'data': serializer.data
        }
        # 返回包含自定义响应数据的 HTTP 201 Created 响应
        return Response(response_data, status=status.HTTP_201_CREATED, headers=headers)

    # 重写 retrieve 方法，该方法用于处理获取单个书籍详细信息的请求
    def retrieve(self, request, *args, **kwargs):
        # 获取要查询的单个书籍实例
        instance = self.get_object()
        # 使用序列化器对该实例进行序列化
        serializer = self.get_serializer(instance)
        # 可以在这里添加额外的逻辑，例如检查权限、增加访问计数等
        # 自定义响应数据的结构，包含状态信息和实际数据
        response_data = {
            'status': 'success',
            'data': serializer.data
        }
        return Response(response_data)

    # 重写 update 方法，该方法用于处理更新书籍信息的请求
    def update(self, request, *args, **kwargs):
        # 判断是否是部分更新，默认为 False
        partial = kwargs.pop('partial', False)
        # 获取要更新的书籍实例
        instance = self.get_object()
        # 使用请求数据和实例创建序列化器实例，partial 表示是否为部分更新
        serializer = self.get_serializer(instance, data=request.data, partial=partial)
        # 验证请求数据的有效性，如果无效则抛出异常
        serializer.is_valid(raise_exception=True)
        # 可以在这里添加额外的逻辑，例如记录更新日志等
        # 调用父类的更新方法更新书籍实例
        self.perform_update(serializer)
        # 如果实例有预取缓存，需要强制使预取缓存失效
        if getattr(instance, '_prefetched_objects_cache', None):
            instance._prefetched_objects_cache = {}
        # 自定义响应数据的结构，包含状态信息、消息和实际数据
        response_data = {
            'status': 'success',
            'message': 'Book updated successfully',
            'data': serializer.data
        }
        return Response(response_data)

    # 重写 destroy 方法，该方法用于处理删除书籍的请求
    def destroy(self, request, *args, **kwargs):
        # 获取要删除的书籍实例
        instance = self.get_object()
        # 可以在这里添加额外的逻辑，例如检查是否可以删除、记录删除日志等
        # 调用父类的删除方法删除书籍实例
        self.perform_destroy(instance)
        # 自定义响应数据的结构，包含状态信息和消息
        response_data = {
            'status': 'success',
            'message': 'Book deleted successfully'
        }
        # 返回包含自定义响应数据的 HTTP 204 No Content 响应
        return Response(response_data, status=status.HTTP_204_NO_CONTENT)

    # 其他常用方法示例：自定义动作
    from rest_framework.decorators import action

    # 使用 @action 装饰器定义一个自定义动作，detail=False 表示该动作不针对单个对象，methods=['get'] 表示只允许 GET 请求
    @action(detail=False, methods=['get'])
    def recent_books(self, request):
        # 获取最近出版的书籍，按出版日期降序排序并取前 5 本
        recent_books = Book.objects.order_by('-publication_date')[:5]
        # 使用序列化器对这些书籍进行序列化，many=True 表示处理多个对象
        serializer = self.get_serializer(recent_books, many=True)
        # 自定义响应数据的结构，包含状态信息和实际数据
        response_data = {
            'status': 'success',
            'data': serializer.data
        }
        return Response(response_data)

2.2、自定义视图逻辑

from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView
from .models import User
from .serializers import UserSerializer
class UserDetail(APIView):
    """自定义用户详情视图（演示APIView用法）"""
    permission_classes = [permissions.IsAuthenticated]
    def get_object(self, pk):
        """封装对象获取逻辑"""
        try:
            return User.objects.get(pk=pk)
        except User.DoesNotExist:
            raise Http404
    
    def get(self, request, pk, format=None):
        """处理GET请求"""
        user = self.get_object(pk)
        self.check_object_permissions(self.request, user)  # 检查对象级权限
        serializer = UserSerializer(user)
        return Response(serializer.data)
    
    def put(self, request, pk, format=None):
        """处理PUT请求（完整更新）"""
        user = self.get_object(pk)
        serializer = UserSerializer(user, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

2.3、异步视图

from asgiref.sync import async_to_sync
from channels.layers import get_channel_layer
from rest_framework import status
from rest_framework.response import Response
from rest_framework.views import APIView
class AsyncNotificationView(APIView):
    """异步通知视图（集成Channels）"""
    authentication_classes = [authentication.TokenAuthentication]  # 使用Token认证
    async def post(self, request, format=None):
        """
        异步POST处理
        功能：
        - 通过WebSocket群组发送通知
        - 演示异步视图写法
        """
        channel_layer = get_channel_layer()
        message = request.data.get("message", "Hello, WebSocket!")
    
        # 发送消息到WebSocket群组
        await channel_layer.group_send(
            "notifications",
            {
                "type": "send.message",
                "message": message,
                "sender": request.user.username  # 添加发送者信息
            }
        )
    
        # 记录审计日志
        AuditLog.objects.create(
            user=request.user,
            action='SEND_NOTIFICATION',
            details=f"Sent message: {message}"
        )
    
        return Response(
            {"status": "notification sent"},
            status=status.HTTP_200_OK
        )

3. 路由（Routers）

3.1、基础路由配置

from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import UserViewSet
router = DefaultRouter(trailing_slash=False)  # 关闭URL结尾斜杠
router.register(r"users", UserViewSet, basename="user")
urlpatterns = [
    path("api/v1/", include((router.urls, 'api'), namespace='v1')),  # 版本化API
    path("api/auth/", include('rest_framework.urls', namespace='rest_framework')),
]

3.2、自定义路由

from rest_framework.routers import Route, DynamicRoute
class CustomRouter(DefaultRouter):
    """
    自定义路由器
    功能：
    - 简化路由配置
    - 增加健康检查端点
    """
    routes = [
        # 自定义列表路由
        Route(
            url=r'^{prefix}/list$',
            mapping={'get': 'list'},
            name='{basename}-list',
            detail=False,
            initkwargs={'suffix': 'List'}
        ),
        # 健康检查路由
        DynamicRoute(
            url=r'^{prefix}/healthz$',
            name='{basename}-health',
            detail=False,
            initkwargs={'action': 'health_check'}
        )
    ]
    def get_api_root_view(self, api_urls=None):
        """重写API根视图"""
        view = super().get_api_root_view(api_urls)
        def wrapper(request, *args, **kwargs):
            response = view(request, *args, **kwargs)
            # 在根API中添加自定义元数据
            response.data['version'] = '1.0.0'
            response.data['environment'] = settings.ENVIRONMENT
            return response
        return wrapper

4、认证与权限

4.1、JWT认证配置示例

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": (
        "rest_framework_simplejwt.authentication.JWTAuthentication",  # JWT认证
        "rest_framework.authentication.SessionAuthentication",        # 会话认证
    ),
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.IsAuthenticatedOrReadOnly",  # 默认权限策略
    ],
    "DEFAULT_THROTTLE_CLASSES": [  # 限流配置
        "rest_framework.throttling.AnonRateThrottle",
        "rest_framework.throttling.UserRateThrottle"
    ],
    "DEFAULT_THROTTLE_RATES": {
        "anon": "100/day",
        "user": "1000/day"
    }
}

4.2、自定义权限类

from rest_framework import permissions
class IsOwnerOrAdmin(permissions.BasePermission):
    """对象级权限：只允许对象所有者或管理员操作"""
    def has_object_permission(self, request, view, obj):
        # 读取权限允许所有请求
        if request.method in permissions.SAFE_METHODS:
            return True
    # 写权限仅限对象所有者或管理员
    return obj.owner == request.user or request.user.is_staff

5、过滤、排序和分页

5.1、复杂过滤示例

from django_filters.rest_framework import DjangoFilterBackend
class AdvancedUserFilter(filters.FilterSet):
    """高级用户过滤器"""
    created_after = filters.DateTimeFilter(
        field_name="date_joined", lookup_expr='gte'
    )
    email_domain = filters.CharFilter(
        method='filter_by_email_domain'
    )
    class Meta:
        model = User
        fields = ['username', 'is_active']
    
    def filter_by_email_domain(self, queryset, name, value):
        """自定义过滤方法：按邮箱域名过滤"""
        return queryset.filter(email__endswith=f"@{value}")
class UserListView(generics.ListAPIView):
    """支持复杂过滤的用户列表视图"""
    queryset = User.objects.all()
    serializer_class = UserSerializer
    filter_backends = [DjangoFilterBackend, filters.OrderingFilter]
    filterset_class = AdvancedUserFilter
    ordering_fields = ['date_joined', 'last_login']

6、补充常用组件

6.1、缓存集成

from django.utils.decorators import method_decorator
from django.views.decorators.cache import cache_page
class CachedUserView(APIView):
    """带缓存功能的用户视图"""
    @method_decorator(cache_page(60*15))  # 缓存15分钟
    def get(self, request, format=None):
        users = User.objects.all()
        serializer = UserSerializer(users, many=True)
        return Response(serializer.data)

6.2、信号处理

from django.db.models.signals import post_save
from django.dispatch import receiver

@receiver(post_save, sender=User)
def user_created_handler(sender, instance, created, **kwargs):
    """用户创建信号处理"""
    if created:
        Profile.objects.create(user=instance)
        # 发送欢迎邮件
        send_mail(
            subject="Welcome!",
            message="Thanks for registering!",
            from_email=settings.DEFAULT_FROM_EMAIL,
            recipient_list=[instance.email]
        )

6.3 中间件示例

class RequestLogMiddleware:
    """请求日志中间件"""
    def init(self, get_response):
        self.get_response = get_response
    def __call__(self, request):
        # 请求处理前
        start_time = time.time()
    
        response = self.get_response(request)
    
        # 请求处理后
        duration = time.time() - start_time
        log_data = {
            "method": request.method,
            "path": request.path,
            "status": response.status_code,
            "duration": duration
        }
        logger.info(f"API Request: {log_data}")
    
        return response

6.4 测试用例

from rest_framework.test import APITestCase
class UserAPITestCase(APITestCase):
    """用户API测试用例"""
    def setUp(self):
        self.user = User.objects.create_user(
            username="testuser",
            password="testpass123"
        )
        self.client.force_authenticate(user=self.user)
    def test_user_list(self):
        response = self.client.get('/api/users/')
        self.assertEqual(response.status_code, 200)
        self.assertEqual(len(response.data), 1)

6.5 性能监控

# settings.py
INSTALLED_APPS += ['django_prometheus']
MIDDLEWARE = [
    'django_prometheus.middleware.PrometheusBeforeMiddleware',
    # ...其他中间件...
    'django_prometheus.middleware.PrometheusAfterMiddleware'
]

6.6、异步任务集成（Celery）

from celery import shared_task

@shared_task
def process_user_data(user_id):
    """后台处理用户数据的Celery任务"""
    user = User.objects.get(id=user_id)
    # 执行耗时操作，如数据分析、文件处理等
    user.data_processed = True
    user.save()
    
    #在视图中调用异步任务
class ProcessUserView(APIView):
    def post(self, request, pk):
        process_user_data.delay(pk)
        return Response({"status": "processing started"})

6.7、数据库读写分离

DATABASE_ROUTERS = ['path.to.PrimaryReplicaRouter']
class PrimaryReplicaRouter:
    """数据库路由：实现读写分离"""
    def db_for_read(self, model, **hints):
        return 'replica'
    def db_for_write(self, model, **hints):
        return 'primary'
    
    def allow_relation(self, obj1, obj2, **hints):
        return True

6.8、API文档增强（drf-yasg）

from drf_yasg import openapi
from drf_yasg.views import get_schema_view
schema_view = get_schema_view(
    openapi.Info(
        title="API Documentation",
        default_version='v1',
        description="详细API文档",
        contact=openapi.Contact(email="support@example.com"),
    ),
    public=True,
)
urlpatterns += [
    path('swagger/', schema_view.with_ui('swagger', cache_timeout=0)),
    path('redoc/', schema_view.with_ui('redoc', cache_timeout=0)),
]

6.9、安全增强

SECURE_CONTENT_TYPE_NOSNIFF = True
X_FRAME_OPTIONS = 'DENY'
CORS_ORIGIN_WHITELIST = [
    'https://example.com',
    'https://api.example.com'
]
CSRF_TRUSTED_ORIGINS = CORS_ORIGIN_WHITELIST

6.10、日志配置

LOGGING = {
    'version': 1,
    'handlers': {
        'file': {
            'level': 'DEBUG',
            'class': 'logging.handlers.TimedRotatingFileHandler',
            'filename': '/var/log/api.log',
            'when': 'midnight',
            'backupCount': 7
        },
    },
    'loggers': {
        'django': {
            'handlers': ['file'],
            'level': 'INFO',
        },
        'api': {
            'handlers': ['file'],
            'level': 'DEBUG',
        }
    }
}

6.11、性能分析中间件

if settings.DEBUG:
    DEBUG_TOOLBAR_CONFIG = {
        'SHOW_TOOLBAR_CALLBACK': lambda request: True,
    }
    INSTALLED_APPS += ['debug_toolbar']
    MIDDLEWARE = ['debug_toolbar.middleware.DebugToolbarMiddleware'] + MIDDLEWARE