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

打印 上一主题 下一主题

主题 883|帖子 883|积分 2649

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对象)。
  1. from rest_framework import serializers
  2. from datetime import datetime
  3. class CustomDateField(serializers.Field):
  4.     """
  5.     自定义日期字段处理类
  6.     功能:
  7.     - 将数据库中的datetime对象格式化为字符串输出
  8.     - 将输入字符串解析为datetime对象
  9.     """
  10.     def to_representation(self, value):
  11.         """将存储的datetime对象转换为字符串"""
  12.         # 使用strftime进行格式化,确保格式统一
  13.         return value.strftime("%Y-%m-%d %H:%M:%S") if value else None
  14.     def to_internal_value(self, data):
  15.         """将输入字符串转换为datetime对象"""
  16.         try:
  17.             # 严格校验日期格式,防止非法输入
  18.             return datetime.strptime(data, "%Y-%m-%d %H:%M:%S")
  19.         except (ValueError, TypeError) as e:
  20.             # 抛出验证错误,会被序列化器的is_valid()捕获
  21.             raise serializers.ValidationError(f"日期格式错误,需要'YYYY-MM-DD HH:MM:SS'格式。错误:{str(e)}")
复制代码
1.1.2、使用已有的字段类型并扩展其功能

偶然可能想要基于现有的字段类型(如serializers.CharField, serializers.IntegerField等)进行扩展。你可以继承这些具体的字段类,并根据需要添加或修改功能。
  1. rom rest_framework import serializers
  2. class PositiveIntegerField(serializers.IntegerField):
  3.     def to_internal_value(self, data):
  4.         value = super().to_internal_value(data)
  5.         if value < 0:
  6.             raise serializers.ValidationError("This field must be a positive integer.")
  7.         return value
  8. class ExampleSerializer(serializers.Serializer):
  9.     id = PositiveIntegerField()
复制代码


  • 界说了一个名为 PositiveIntegerField 的类,继承自 IntegerField。
  • IntegerField 是 Django REST Framework 提供的标准字段,用于处理整数类型的验证。
  • to_internal_value 是 DRF 中用于将输入数据(如前端传入的值)转换为内部表示的方法。
  • 在序列化器中,to_internal_value 方法用于对输入数据进行验证和转换。
  1. value = super().to_internal_value(data)
复制代码


  • 调用 IntegerField 的 to_internal_value 方法,先对输入数据进行整数类型的验证。
  • 如果输入数据不是整数,会在此处抛出验证错误。
  1. if value < 0:
  2.     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。这允许你在序列化器中界说一个方法,该方法将被用来生成字段值。
  1. from rest_framework import serializers
  2. class UserSerializer(serializers.Serializer):
  3.     username = serializers.CharField()
  4.     email = serializers.EmailField()
  5.     full_name = serializers.SerializerMethodField()  # 只读字段
  6.     def get_full_name(self, obj):
  7.         """
  8.         自定义方法,根据对象生成全名。
  9.         """
  10.         return f"{obj.first_name} {obj.last_name}"
复制代码
请注意,在上面的例子中,get_full_name方法会接收序列化过程中的对象实例作为参数,并返回盼望的值。这个方法的名字必须遵循get_<field_name>的定名规则。
1.1.4、使用**serializers.Serializer**作为字段

  1. from rest_framework import serializers
  2. class AddressSerializer(serializers.Serializer):
  3.     street = serializers.CharField()
  4.     city = serializers.CharField()
  5.     state = serializers.CharField()
  6. class UserSerializer(serializers.Serializer):
  7.     name = serializers.CharField()
  8.     email = serializers.EmailField()
  9.     address = AddressSerializer()  # 使用另一个Serializer作为字段
复制代码
1.1.5、格式化字段(datetime)

  1. class LoginlogSerializer(serializers.ModelSerializer):
  2.     login_time=serializers.DateTimeField(format='%Y年%m月%d日 %H:%M:%S',read_only=True)
  3.     class Meta:
  4.         model=XCLoginlog
  5.         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 嵌套序列化器

  1. from rest_framework import serializers
  2. from .models import User, Profile
  3. class ProfileSerializer(serializers.ModelSerializer):
  4.     """用户资料序列化器,用于将Profile模型实例与Python原生数据类型之间进行转换"""
  5.     class Meta:
  6.         # 指定关联的模型为Profile
  7.         model = Profile
  8.         # 定义序列化和反序列化时包含的字段
  9.         fields = ["bio", "website"]
  10.         # 扩展配置:设置字段额外属性
  11.         extra_kwargs = {
  12.             # 允许bio字段为空字符串,且该字段不是必需的
  13.             'bio': {'allow_blank': True, 'required': False},
  14.             # 为website字段添加URL验证器,确保输入的是有效的URL
  15.             'website': {'validators': [serializers.URLValidator()]}
  16.         }
  17. class UserSerializer(serializers.ModelSerializer):
  18.     """用户主序列化器,用于将User模型实例与Python原生数据类型之间进行转换"""
  19.     # 将ProfileSerializer作为嵌套序列化器,设置为非必须字段以支持部分更新
  20.     profile = ProfileSerializer(required=False)
  21.     class Meta:
  22.         # 指定关联的模型为User
  23.         model = User
  24.         # 定义序列化和反序列化时包含的字段
  25.         fields = ["username", "email", "profile"]
  26.         # 扩展配置:设置字段验证规则
  27.         extra_kwargs = {
  28.             'email': {
  29.                 # 邮箱字段是必需的
  30.                 'required': True,
  31.                 # 为邮箱字段添加唯一验证器,确保该邮箱在所有用户中是唯一的
  32.                 'validators': [
  33.                     serializers.UniqueValidator(
  34.                         queryset=User.objects.all(),
  35.                         message="该邮箱已被注册"
  36.                     )
  37.                 ]
  38.             }
  39.         }
  40.     def create(self, validated_data):
  41.         """
  42.         重写创建方法以处理嵌套关系
  43.         当使用该序列化器创建新的User实例时,此方法会被调用
  44.         """
  45.         # 从验证后的数据中弹出profile数据,如果不存在则返回None
  46.         profile_data = validated_data.pop('profile', None)
  47.         # 使用剩余的验证数据创建User实例
  48.         user = User.objects.create(**validated_data)
  49.         # 如果存在profile数据
  50.         if profile_data:
  51.             # 创建与该用户关联的Profile实例
  52.             Profile.objects.create(user=user, **profile_data)
  53.         # 返回创建好的User实例
  54.         return user
  55.     def update(self, instance, validated_data):
  56.         """
  57.         重写更新方法以处理嵌套关系
  58.         当使用该序列化器更新已有的User实例时,此方法会被调用
  59.         """
  60.         # 从验证后的数据中弹出profile数据,如果不存在则返回空字典
  61.         profile_data = validated_data.pop('profile', {})
  62.         # 调用父类的update方法更新用户实例
  63.         instance = super().update(instance, validated_data)
  64.         # 获取用户关联的Profile实例
  65.         profile = instance.profile
  66.         # 如果用户已经有关联的Profile实例
  67.         if profile:
  68.             # 遍历profile数据中的每个属性和值
  69.             for attr, value in profile_data.items():
  70.                 # 为Profile实例设置新的属性值
  71.                 setattr(profile, attr, value)
  72.             # 保存更新后的Profile实例
  73.             profile.save()
  74.         else:
  75.             # 如果用户没有关联的Profile实例,则创建一个新的Profile实例并关联到该用户
  76.             Profile.objects.create(user=instance, **profile_data)
  77.         # 返回更新后的User实例
  78.         return instance
复制代码
本身编写代码的案例,权限管理体系中菜单序列化
  1. from rest_framework import serializers
  2. from userapi.models import XCMenu,XCRole
  3. class MenuSerializer(serializers.ModelSerializer):
  4.     create_time=serializers.DateTimeField(format='%Y年%m月%d日 %H:%M:%S',read_only=True)
  5.     update_time=serializers.DateTimeField(format='%Y年%m月%d日 %H:%M:%S',read_only=True)
  6.     creator = serializers.SerializerMethodField()
  7.     children = serializers.SerializerMethodField()
  8.     is_top_level = serializers.SerializerMethodField()
  9.     class Meta:
  10.         model = XCMenu
  11.         fields = ['id','name','name_zh','create_time','update_time','creator','path','redrect','component','meta','status',"is_top_level",'parent','children']
  12.     def get_creator(self, obj):
  13.         # 返回 creator 的 username 字段
  14.         return obj.creator.username if obj.creator else None
  15.     def get_children(self, obj):
  16.         # 递归获取子菜单
  17.         # 获取当前菜单的子菜单
  18.         children = XCMenu.objects.filter(parent=obj)
  19.         # 使用 MenuSerializer 序列化子菜单
  20.         return MenuSerializer(children, many=True).data
  21.     def get_is_top_level(self, obj):
  22.         # 如果菜单没有父菜单,则为顶级菜单
  23.         return obj.parent is None
复制代码
1.3 动态字段

  1. class DynamicFieldsModelSerializer(serializers.ModelSerializer):
  2.     """
  3.     支持动态字段控制的序列化器基类
  4.     功能:
  5.     - 允许通过fields参数指定需要的字段
  6.     - 允许通过exclude参数指定排除的字段
  7.     """
  8.     def init(self, *args, **kwargs):
  9.         # 从参数中提取字段控制参数
  10.         fields = kwargs.pop('fields', None)
  11.         exclude = kwargs.pop('exclude', None)
  12.         super().__init__(*args, **kwargs)
  13.         
  14.         if fields is not None and exclude is not None:
  15.             raise serializers.ValidationError("不能同时指定fields和exclude参数")
  16.         
  17.             if fields is not None:
  18.                 # 白名单模式:只保留指定字段
  19.                 allowed = set(fields)
  20.                 existing = set(self.fields)
  21.                 for field_name in existing - allowed:
  22.                     self.fields.pop(field_name)
  23.    
  24.         if exclude is not None:
  25.             # 黑名单模式:排除指定字段
  26.             excluded = set(exclude)
  27.             for field_name in excluded:
  28.                 self.fields.pop(field_name, None)  # 安全删除字段
复制代码
2. 视图(Views)

2.1、视图集(ViewSets)

  1. from rest_framework import viewsets, permissions
  2. from .models import User
  3. from .serializers import UserSerializer
  4. class UserViewSet(viewsets.ModelViewSet):
  5.     """
  6.     用户视图集(完整CRUD)
  7.     功能:
  8.     - 自动生成标准REST接口
  9.     - 包含列表、详情、创建、更新、删除操作
  10.     """
  11.     queryset = User.objects.all().select_related('profile')  # 优化查询性能
  12.     serializer_class = UserSerializer
  13.     permission_classes = [permissions.IsAuthenticatedOrReadOnly]  # 认证用户可写,匿名用户只读
  14.     filter_backends = [filters.OrderingFilter]  # 启用排序
  15.     ordering_fields = ['date_joined', 'username']  # 允许排序的字段
  16.    
  17.     def get_queryset(self):
  18.         """重写查询集以实现业务逻辑过滤"""
  19.         queryset = super().get_queryset()
  20.         # 示例:根据请求参数过滤活跃用户
  21.         is_active = self.request.query_params.get('active', None)
  22.         if is_active in ['true', 'false']:
  23.             queryset = queryset.filter(is_active=is_active.lower() == 'true')
  24.         return queryset
复制代码
重写视图基

  • **list** 方法:用于获取所有书籍的列表。重写该方法时,我们可以添加额外的逻辑,如过滤、排序等,并对返回的数据进行额外处理。
  • **create** 方法:用于创建新的书籍。重写该方法时,我们可以添加额外的逻辑,如记载日记、发送通知等。
  • **retrieve** 方法:用于获取单个书籍的详细信息。重写该方法时,我们可以添加额外的逻辑,如检查权限、增长访问计数等。
  • **update** 方法:用于更新书籍的信息。重写该方法时,我们可以添加额外的逻辑,如记载更新日记等。
  • **destroy** 方法:用于删除书籍。重写该方法时,我们可以添加额外的逻辑,如检查是否可以删除、记载删除日记等。
  • **recent_books** 方法:这是一个自界说动作,用于获取近来出版的书籍。使用 @action 装饰器可以将自界说动作添加到视图会合。
  1. from rest_framework import viewsets
  2. from .models import Book
  3. from .serializers import BookSerializer
  4. from rest_framework.response import Response
  5. from rest_framework import status
  6. class BookViewSet(viewsets.ModelViewSet):
  7.     # 指定视图集要处理的查询集,这里获取所有的 Book 模型实例
  8.     queryset = Book.objects.all()
  9.     # 指定用于序列化和反序列化的序列化器类
  10.     serializer_class = BookSerializer
  11.     # 重写 list 方法,该方法用于处理获取所有书籍列表的请求
  12.     def list(self, request, *args, **kwargs):
  13.         # 可以在这里添加额外的逻辑,例如过滤、排序等
  14.         # 调用父类的过滤方法对查询集进行过滤
  15.         queryset = self.filter_queryset(self.get_queryset())
  16.         # 对过滤后的查询集进行分页处理
  17.         page = self.paginate_queryset(queryset)
  18.         # 如果分页结果不为空
  19.         if page is not None:
  20.             # 使用序列化器对分页后的结果进行序列化,many=True 表示处理多个对象
  21.             serializer = self.get_serializer(page, many=True)
  22.             # 返回分页后的响应数据
  23.             return self.get_paginated_response(serializer.data)
  24.         # 如果没有进行分页,直接对查询集进行序列化
  25.         serializer = self.get_serializer(queryset, many=True)
  26.         # 可以在这里对返回的数据进行额外处理
  27.         # 自定义响应数据的结构,包含状态信息和实际数据
  28.         response_data = {
  29.             'status': 'success',
  30.             'data': serializer.data
  31.         }
  32.         return Response(response_data)
  33.     # 重写 create 方法,该方法用于处理创建新书籍的请求
  34.     def create(self, request, *args, **kwargs):
  35.         # 使用请求数据创建序列化器实例
  36.         serializer = self.get_serializer(data=request.data)
  37.         # 验证请求数据的有效性,如果无效则抛出异常
  38.         serializer.is_valid(raise_exception=True)
  39.         # 可以在这里添加额外的逻辑,例如记录日志、发送通知等
  40.         # 调用父类的创建方法创建新的书籍实例
  41.         self.perform_create(serializer)
  42.         # 获取成功响应的头部信息
  43.         headers = self.get_success_headers(serializer.data)
  44.         # 自定义响应数据的结构,包含状态信息、消息和实际数据
  45.         response_data = {
  46.             'status': 'success',
  47.             'message': 'Book created successfully',
  48.             'data': serializer.data
  49.         }
  50.         # 返回包含自定义响应数据的 HTTP 201 Created 响应
  51.         return Response(response_data, status=status.HTTP_201_CREATED, headers=headers)
  52.     # 重写 retrieve 方法,该方法用于处理获取单个书籍详细信息的请求
  53.     def retrieve(self, request, *args, **kwargs):
  54.         # 获取要查询的单个书籍实例
  55.         instance = self.get_object()
  56.         # 使用序列化器对该实例进行序列化
  57.         serializer = self.get_serializer(instance)
  58.         # 可以在这里添加额外的逻辑,例如检查权限、增加访问计数等
  59.         # 自定义响应数据的结构,包含状态信息和实际数据
  60.         response_data = {
  61.             'status': 'success',
  62.             'data': serializer.data
  63.         }
  64.         return Response(response_data)
  65.     # 重写 update 方法,该方法用于处理更新书籍信息的请求
  66.     def update(self, request, *args, **kwargs):
  67.         # 判断是否是部分更新,默认为 False
  68.         partial = kwargs.pop('partial', False)
  69.         # 获取要更新的书籍实例
  70.         instance = self.get_object()
  71.         # 使用请求数据和实例创建序列化器实例,partial 表示是否为部分更新
  72.         serializer = self.get_serializer(instance, data=request.data, partial=partial)
  73.         # 验证请求数据的有效性,如果无效则抛出异常
  74.         serializer.is_valid(raise_exception=True)
  75.         # 可以在这里添加额外的逻辑,例如记录更新日志等
  76.         # 调用父类的更新方法更新书籍实例
  77.         self.perform_update(serializer)
  78.         # 如果实例有预取缓存,需要强制使预取缓存失效
  79.         if getattr(instance, '_prefetched_objects_cache', None):
  80.             instance._prefetched_objects_cache = {}
  81.         # 自定义响应数据的结构,包含状态信息、消息和实际数据
  82.         response_data = {
  83.             'status': 'success',
  84.             'message': 'Book updated successfully',
  85.             'data': serializer.data
  86.         }
  87.         return Response(response_data)
  88.     # 重写 destroy 方法,该方法用于处理删除书籍的请求
  89.     def destroy(self, request, *args, **kwargs):
  90.         # 获取要删除的书籍实例
  91.         instance = self.get_object()
  92.         # 可以在这里添加额外的逻辑,例如检查是否可以删除、记录删除日志等
  93.         # 调用父类的删除方法删除书籍实例
  94.         self.perform_destroy(instance)
  95.         # 自定义响应数据的结构,包含状态信息和消息
  96.         response_data = {
  97.             'status': 'success',
  98.             'message': 'Book deleted successfully'
  99.         }
  100.         # 返回包含自定义响应数据的 HTTP 204 No Content 响应
  101.         return Response(response_data, status=status.HTTP_204_NO_CONTENT)
  102.     # 其他常用方法示例:自定义动作
  103.     from rest_framework.decorators import action
  104.     # 使用 @action 装饰器定义一个自定义动作,detail=False 表示该动作不针对单个对象,methods=['get'] 表示只允许 GET 请求
  105.     @action(detail=False, methods=['get'])
  106.     def recent_books(self, request):
  107.         # 获取最近出版的书籍,按出版日期降序排序并取前 5 本
  108.         recent_books = Book.objects.order_by('-publication_date')[:5]
  109.         # 使用序列化器对这些书籍进行序列化,many=True 表示处理多个对象
  110.         serializer = self.get_serializer(recent_books, many=True)
  111.         # 自定义响应数据的结构,包含状态信息和实际数据
  112.         response_data = {
  113.             'status': 'success',
  114.             'data': serializer.data
  115.         }
  116.         return Response(response_data)
复制代码
2.2、自界说视图逻辑

  1. from rest_framework import status
  2. from rest_framework.response import Response
  3. from rest_framework.views import APIView
  4. from .models import User
  5. from .serializers import UserSerializer
  6. class UserDetail(APIView):
  7.     """自定义用户详情视图(演示APIView用法)"""
  8.     permission_classes = [permissions.IsAuthenticated]
  9.     def get_object(self, pk):
  10.         """封装对象获取逻辑"""
  11.         try:
  12.             return User.objects.get(pk=pk)
  13.         except User.DoesNotExist:
  14.             raise Http404
  15.    
  16.     def get(self, request, pk, format=None):
  17.         """处理GET请求"""
  18.         user = self.get_object(pk)
  19.         self.check_object_permissions(self.request, user)  # 检查对象级权限
  20.         serializer = UserSerializer(user)
  21.         return Response(serializer.data)
  22.    
  23.     def put(self, request, pk, format=None):
  24.         """处理PUT请求(完整更新)"""
  25.         user = self.get_object(pk)
  26.         serializer = UserSerializer(user, data=request.data)
  27.         if serializer.is_valid():
  28.             serializer.save()
  29.             return Response(serializer.data)
  30.         return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
复制代码
2.3、异步视图

  1. from asgiref.sync import async_to_sync
  2. from channels.layers import get_channel_layer
  3. from rest_framework import status
  4. from rest_framework.response import Response
  5. from rest_framework.views import APIView
  6. class AsyncNotificationView(APIView):
  7.     """异步通知视图(集成Channels)"""
  8.     authentication_classes = [authentication.TokenAuthentication]  # 使用Token认证
  9.     async def post(self, request, format=None):
  10.         """
  11.         异步POST处理
  12.         功能:
  13.         - 通过WebSocket群组发送通知
  14.         - 演示异步视图写法
  15.         """
  16.         channel_layer = get_channel_layer()
  17.         message = request.data.get("message", "Hello, WebSocket!")
  18.    
  19.         # 发送消息到WebSocket群组
  20.         await channel_layer.group_send(
  21.             "notifications",
  22.             {
  23.                 "type": "send.message",
  24.                 "message": message,
  25.                 "sender": request.user.username  # 添加发送者信息
  26.             }
  27.         )
  28.    
  29.         # 记录审计日志
  30.         AuditLog.objects.create(
  31.             user=request.user,
  32.             action='SEND_NOTIFICATION',
  33.             details=f"Sent message: {message}"
  34.         )
  35.    
  36.         return Response(
  37.             {"status": "notification sent"},
  38.             status=status.HTTP_200_OK
  39.         )
复制代码
3. 路由(Routers)

3.1、基础路由设置

  1. from django.urls import path, include
  2. from rest_framework.routers import DefaultRouter
  3. from .views import UserViewSet
  4. router = DefaultRouter(trailing_slash=False)  # 关闭URL结尾斜杠
  5. router.register(r"users", UserViewSet, basename="user")
  6. urlpatterns = [
  7.     path("api/v1/", include((router.urls, 'api'), namespace='v1')),  # 版本化API
  8.     path("api/auth/", include('rest_framework.urls', namespace='rest_framework')),
  9. ]
复制代码
3.2、自界说路由

  1. from rest_framework.routers import Route, DynamicRoute
  2. class CustomRouter(DefaultRouter):
  3.     """
  4.     自定义路由器
  5.     功能:
  6.     - 简化路由配置
  7.     - 增加健康检查端点
  8.     """
  9.     routes = [
  10.         # 自定义列表路由
  11.         Route(
  12.             url=r'^{prefix}/list$',
  13.             mapping={'get': 'list'},
  14.             name='{basename}-list',
  15.             detail=False,
  16.             initkwargs={'suffix': 'List'}
  17.         ),
  18.         # 健康检查路由
  19.         DynamicRoute(
  20.             url=r'^{prefix}/healthz$',
  21.             name='{basename}-health',
  22.             detail=False,
  23.             initkwargs={'action': 'health_check'}
  24.         )
  25.     ]
  26.     def get_api_root_view(self, api_urls=None):
  27.         """重写API根视图"""
  28.         view = super().get_api_root_view(api_urls)
  29.         def wrapper(request, *args, **kwargs):
  30.             response = view(request, *args, **kwargs)
  31.             # 在根API中添加自定义元数据
  32.             response.data['version'] = '1.0.0'
  33.             response.data['environment'] = settings.ENVIRONMENT
  34.             return response
  35.         return wrapper
复制代码
4、认证与权限

4.1、JWT认证设置示例

  1. REST_FRAMEWORK = {
  2.     "DEFAULT_AUTHENTICATION_CLASSES": (
  3.         "rest_framework_simplejwt.authentication.JWTAuthentication",  # JWT认证
  4.         "rest_framework.authentication.SessionAuthentication",        # 会话认证
  5.     ),
  6.     "DEFAULT_PERMISSION_CLASSES": [
  7.         "rest_framework.permissions.IsAuthenticatedOrReadOnly",  # 默认权限策略
  8.     ],
  9.     "DEFAULT_THROTTLE_CLASSES": [  # 限流配置
  10.         "rest_framework.throttling.AnonRateThrottle",
  11.         "rest_framework.throttling.UserRateThrottle"
  12.     ],
  13.     "DEFAULT_THROTTLE_RATES": {
  14.         "anon": "100/day",
  15.         "user": "1000/day"
  16.     }
  17. }
复制代码
4.2、自界说权限类

  1. from rest_framework import permissions
  2. class IsOwnerOrAdmin(permissions.BasePermission):
  3.     """对象级权限:只允许对象所有者或管理员操作"""
  4.     def has_object_permission(self, request, view, obj):
  5.         # 读取权限允许所有请求
  6.         if request.method in permissions.SAFE_METHODS:
  7.             return True
  8.     # 写权限仅限对象所有者或管理员
  9.     return obj.owner == request.user or request.user.is_staff
复制代码
5、过滤、排序和分页

5.1、复杂过滤示例

  1. from django_filters.rest_framework import DjangoFilterBackend
  2. class AdvancedUserFilter(filters.FilterSet):
  3.     """高级用户过滤器"""
  4.     created_after = filters.DateTimeFilter(
  5.         field_name="date_joined", lookup_expr='gte'
  6.     )
  7.     email_domain = filters.CharFilter(
  8.         method='filter_by_email_domain'
  9.     )
  10.     class Meta:
  11.         model = User
  12.         fields = ['username', 'is_active']
  13.    
  14.     def filter_by_email_domain(self, queryset, name, value):
  15.         """自定义过滤方法:按邮箱域名过滤"""
  16.         return queryset.filter(email__endswith=f"@{value}")
  17. class UserListView(generics.ListAPIView):
  18.     """支持复杂过滤的用户列表视图"""
  19.     queryset = User.objects.all()
  20.     serializer_class = UserSerializer
  21.     filter_backends = [DjangoFilterBackend, filters.OrderingFilter]
  22.     filterset_class = AdvancedUserFilter
  23.     ordering_fields = ['date_joined', 'last_login']
复制代码
6、补充常用组件

6.1、缓存集成

  1. from django.utils.decorators import method_decorator
  2. from django.views.decorators.cache import cache_page
  3. class CachedUserView(APIView):
  4.     """带缓存功能的用户视图"""
  5.     @method_decorator(cache_page(60*15))  # 缓存15分钟
  6.     def get(self, request, format=None):
  7.         users = User.objects.all()
  8.         serializer = UserSerializer(users, many=True)
  9.         return Response(serializer.data)
复制代码
6.2、信号处理

  1. from django.db.models.signals import post_save
  2. from django.dispatch import receiver
  3. @receiver(post_save, sender=User)
  4. def user_created_handler(sender, instance, created, **kwargs):
  5.     """用户创建信号处理"""
  6.     if created:
  7.         Profile.objects.create(user=instance)
  8.         # 发送欢迎邮件
  9.         send_mail(
  10.             subject="Welcome!",
  11.             message="Thanks for registering!",
  12.             from_email=settings.DEFAULT_FROM_EMAIL,
  13.             recipient_list=[instance.email]
  14.         )
复制代码
6.3 中央件示例

  1. class RequestLogMiddleware:
  2.     """请求日志中间件"""
  3.     def init(self, get_response):
  4.         self.get_response = get_response
  5.     def __call__(self, request):
  6.         # 请求处理前
  7.         start_time = time.time()
  8.    
  9.         response = self.get_response(request)
  10.    
  11.         # 请求处理后
  12.         duration = time.time() - start_time
  13.         log_data = {
  14.             "method": request.method,
  15.             "path": request.path,
  16.             "status": response.status_code,
  17.             "duration": duration
  18.         }
  19.         logger.info(f"API Request: {log_data}")
  20.    
  21.         return response
复制代码
6.4 测试用例

  1. from rest_framework.test import APITestCase
  2. class UserAPITestCase(APITestCase):
  3.     """用户API测试用例"""
  4.     def setUp(self):
  5.         self.user = User.objects.create_user(
  6.             username="testuser",
  7.             password="testpass123"
  8.         )
  9.         self.client.force_authenticate(user=self.user)
  10.     def test_user_list(self):
  11.         response = self.client.get('/api/users/')
  12.         self.assertEqual(response.status_code, 200)
  13.         self.assertEqual(len(response.data), 1)
复制代码
6.5 性能监控

  1. # settings.py
  2. INSTALLED_APPS += ['django_prometheus']
  3. MIDDLEWARE = [
  4.     'django_prometheus.middleware.PrometheusBeforeMiddleware',
  5.     # ...其他中间件...
  6.     'django_prometheus.middleware.PrometheusAfterMiddleware'
  7. ]
复制代码
6.6、异步任务集成(Celery)

  1. from celery import shared_task
  2. @shared_task
  3. def process_user_data(user_id):
  4.     """后台处理用户数据的Celery任务"""
  5.     user = User.objects.get(id=user_id)
  6.     # 执行耗时操作,如数据分析、文件处理等
  7.     user.data_processed = True
  8.     user.save()
  9.    
  10.     #在视图中调用异步任务
  11. class ProcessUserView(APIView):
  12.     def post(self, request, pk):
  13.         process_user_data.delay(pk)
  14.         return Response({"status": "processing started"})
复制代码


6.7、数据库读写分离

  1. DATABASE_ROUTERS = ['path.to.PrimaryReplicaRouter']
  2. class PrimaryReplicaRouter:
  3.     """数据库路由:实现读写分离"""
  4.     def db_for_read(self, model, **hints):
  5.         return 'replica'
  6.     def db_for_write(self, model, **hints):
  7.         return 'primary'
  8.    
  9.     def allow_relation(self, obj1, obj2, **hints):
  10.         return True
复制代码
6.8、API文档增强(drf-yasg)

  1. from drf_yasg import openapi
  2. from drf_yasg.views import get_schema_view
  3. schema_view = get_schema_view(
  4.     openapi.Info(
  5.         title="API Documentation",
  6.         default_version='v1',
  7.         description="详细API文档",
  8.         contact=openapi.Contact(email="support@example.com"),
  9.     ),
  10.     public=True,
  11. )
  12. urlpatterns += [
  13.     path('swagger/', schema_view.with_ui('swagger', cache_timeout=0)),
  14.     path('redoc/', schema_view.with_ui('redoc', cache_timeout=0)),
  15. ]
复制代码
6.9、安全增强

  1. SECURE_CONTENT_TYPE_NOSNIFF = True
  2. X_FRAME_OPTIONS = 'DENY'
  3. CORS_ORIGIN_WHITELIST = [
  4.     'https://example.com',
  5.     'https://api.example.com'
  6. ]
  7. CSRF_TRUSTED_ORIGINS = CORS_ORIGIN_WHITELIST
复制代码
6.10、日记设置

  1. LOGGING = {
  2.     'version': 1,
  3.     'handlers': {
  4.         'file': {
  5.             'level': 'DEBUG',
  6.             'class': 'logging.handlers.TimedRotatingFileHandler',
  7.             'filename': '/var/log/api.log',
  8.             'when': 'midnight',
  9.             'backupCount': 7
  10.         },
  11.     },
  12.     'loggers': {
  13.         'django': {
  14.             'handlers': ['file'],
  15.             'level': 'INFO',
  16.         },
  17.         'api': {
  18.             'handlers': ['file'],
  19.             'level': 'DEBUG',
  20.         }
  21.     }
  22. }
复制代码
6.11、性能分析中央件

  1. if settings.DEBUG:
  2.     DEBUG_TOOLBAR_CONFIG = {
  3.         'SHOW_TOOLBAR_CALLBACK': lambda request: True,
  4.     }
  5.     INSTALLED_APPS += ['debug_toolbar']
  6.     MIDDLEWARE = ['debug_toolbar.middleware.DebugToolbarMiddleware'] + MIDDLEWARE
复制代码
免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。
回复

使用道具 举报

0 个回复

倒序浏览

快速回复

您需要登录后才可以回帖 登录 or 立即注册

本版积分规则

铁佛

金牌会员
这个人很懒什么都没写!
快速回复 返回顶部 返回列表