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 Nonedef 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 serializersclass 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 serializersclass 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 serializersclass 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=XCLoginlogfields=('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, Profileclass ProfileSerializer(serializers.ModelSerializer):"""用户资料序列化器，用于将Profile模型实例与Python原生数据类型之间进行转换"""class Meta:# 指定关联的模型为Profilemodel = 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:# 指定关联的模型为Usermodel = 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数据，如果不存在则返回Noneprofile_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 userdef 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,XCRoleclass 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 = XCMenufields = ['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 Nonedef get_children(self, obj):# 递归获取子菜单# 获取当前菜单的子菜单children = XCMenu.objects.filter(parent=obj)# 使用 MenuSerializer 序列化子菜单return MenuSerializer(children, many=True).datadef 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 = UserSerializerpermission_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 statusclass 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):# 判断是否是部分更新，默认为 Falsepartial = 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 Http404def 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')), # 版本化APIpath("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.ENVIRONMENTreturn responsereturn 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 = Userfields = ['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 = UserSerializerfilter_backends = [DjangoFilterBackend, filters.OrderingFilter]filterset_class = AdvancedUserFilterordering_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_responsedef __call__(self, request):# 请求处理前start_time = time.time()response = self.get_response(request)# 请求处理后duration = time.time() - start_timelog_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 = Trueuser.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