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
复制代码 重写视图基
- **list** 方法:用于获取所有书籍的列表。重写该方法时,我们可以添加额外的逻辑,如过滤、排序等,并对返回的数据进行额外处理。
- **create** 方法:用于创建新的书籍。重写该方法时,我们可以添加额外的逻辑,如记载日记、发送通知等。
- **retrieve** 方法:用于获取单个书籍的详细信息。重写该方法时,我们可以添加额外的逻辑,如检查权限、增长访问计数等。
- **update** 方法:用于更新书籍的信息。重写该方法时,我们可以添加额外的逻辑,如记载更新日记等。
- **destroy** 方法:用于删除书籍。重写该方法时,我们可以添加额外的逻辑,如检查是否可以删除、记载删除日记等。
- **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
复制代码 免责声明:如果侵犯了您的权益,请联系站长,我们会及时删除侵权内容,谢谢合作!更多信息从访问主页:qidao123.com:ToB企服之家,中国第一个企服评测及商务社交产业平台。 |