安装与配置

安装

bash
pip install djangorestframework
pip install markdown       # 可选，用于支持markdown
pip install django-filter  # 可选，用于过滤支持

基础配置

settings.py
INSTALLED_APPS = [
    ...
    'rest_framework',
    'django_filters',  # 可选
]

REST_FRAMEWORK = {
    # 默认权限配置
    'DEFAULT_PERMISSION_CLASSES': [
        'rest_framework.permissions.IsAuthenticated',
    ],
    
    # 默认认证配置
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.SessionAuthentication',
        'rest_framework.authentication.TokenAuthentication',
    ],
    
    # 分页配置
    'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
    'PAGE_SIZE': 20,
    
    # 解析器配置
    'DEFAULT_PARSER_CLASSES': [
        'rest_framework.parsers.JSONParser',
        'rest_framework.parsers.FormParser',
        'rest_framework.parsers.MultiPartParser',
    ],
    
    # 渲染器配置
    'DEFAULT_RENDERER_CLASSES': [
        'rest_framework.renderers.JSONRenderer',
        'rest_framework.renderers.BrowsableAPIRenderer',
    ],
    
    # 限流配置
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle'
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',
        'user': '1000/day'
    },
    
    # 版本控制
    'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.URLPathVersioning',
}

URL配置

urls.py
from django.urls import path, include

urlpatterns = [
    ...
    path('api-auth/', include('rest_framework.urls')),  # 登录注销URL
]

5分钟快速入门

1. 创建模型

models.py
from django.db import models

class Book(models.Model):
    title = models.CharField(max_length=100)
    author = models.CharField(max_length=100)
    price = models.DecimalField(max_digits=10, decimal_places=2)
    published_date = models.DateField()
    is_available = models.BooleanField(default=True)
    
    class Meta:
        db_table = 'books'
        
    def __str__(self):
        return self.title

2. 创建序列化器

serializers.py
from rest_framework import serializers
from .models import Book

class BookSerializer(serializers.ModelSerializer):
    class Meta:
        model = Book
        fields = ['id', 'title', 'author', 'price', 'published_date', 'is_available']
        read_only_fields = ['id']  # id字段只读

3. 创建视图

views.py
from rest_framework import viewsets
from .models import Book
from .serializers import BookSerializer

class BookViewSet(viewsets.ModelViewSet):
    """
    提供Book的CRUD操作
    """
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    # 可选：自定义权限
    # permission_classes = [IsAuthenticated]
    
    # 可选：自定义查询集
    def get_queryset(self):
        queryset = Book.objects.all()
        is_available = self.request.query_params.get('is_available')
        if is_available is not None:
            queryset = queryset.filter(is_available=is_available)
        return queryset

4. 配置路由

urls.py
from django.urls import path, include
from rest_framework.routers import DefaultRouter
from .views import BookViewSet

router = DefaultRouter()
router.register(r'books', BookViewSet)

urlpatterns = [
    path('', include(router.urls)),
]
完成！ 现在你可以访问：
• GET/POST /api/books/ - 列表和创建
• GET/PUT/DELETE /api/books/1/ - 详情、更新、删除

项目结构建议

小型项目结构

myproject/
├── apps/
│   ├── __init__.py
│   ├── users/
│   │   ├── __init__.py
│   │   ├── models.py
│   │   ├── serializers.py
│   │   ├── views.py
│   │   ├── urls.py
│   │   └── tests.py
│   └── products/
│       ├── __init__.py
│       ├── models.py
│       ├── serializers.py
│       ├── views.py
│       └── urls.py
├── config/
│   ├── __init__.py
│   ├── settings.py
│   ├── urls.py
│   └── wsgi.py
├── utils/
│   ├── __init__.py
│   ├── permissions.py
│   └── authentication.py
├── manage.py
└── requirements.txt

中大型项目结构

myproject/
├── apps/
│   ├── __init__.py
│   ├── common/              # 公共组件
│   │   ├── __init__.py
│   │   ├── models.py        # 抽象基类
│   │   ├── serializers.py   # 通用序列化器
│   │   └── views.py         # 通用视图
│   ├── users/
│   │   ├── __init__.py
│   │   ├── models/
│   │   │   ├── __init__.py
│   │   │   └── user.py
│   │   ├── serializers/
│   │   │   ├── __init__.py
│   │   │   ├── user.py
│   │   │   └── auth.py
│   │   ├── views/
│   │   │   ├── __init__.py
│   │   │   ├── user.py
│   │   │   └── auth.py
│   │   ├── urls.py
│   │   └── tests/
│   └── products/
│       └── ...
├── config/
├── utils/
├── docs/
├── scripts/
└── tests/

序列化器 (Serializers)

基础概念

序列化器负责将复杂数据（如QuerySet和模型实例）转换为Python原生数据类型，以便渲染为JSON、XML等格式。同时也负责反序列化（验证和转换传入数据）。

Serializer vs ModelSerializer

特性 Serializer ModelSerializer
字段定义 手动定义每个字段 自动从模型生成
create/update 手动实现 自动生成
灵活性 高 中
适用场景 非模型数据、复杂逻辑 标准CRUD操作

基础Serializer示例

serializers.py
from rest_framework import serializers

class CommentSerializer(serializers.Serializer):
    id = serializers.IntegerField(read_only=True)
    title = serializers.CharField(max_length=100)
    content = serializers.CharField(max_length=1000)
    created = serializers.DateTimeField(read_only=True)
    email = serializers.EmailField()
    
    def create(self, validated_data):
        """
        创建新实例
        """
        return Comment.objects.create(**validated_data)
    
    def update(self, instance, validated_data):
        """
        更新现有实例
        """
        instance.title = validated_data.get('title', instance.title)
        instance.content = validated_data.get('content', instance.content)
        instance.email = validated_data.get('email', instance.email)
        instance.save()
        return instance

ModelSerializer高级用法

serializers.py
from rest_framework import serializers
from .models import Book, Author

class AuthorSerializer(serializers.ModelSerializer):
    class Meta:
        model = Author
        fields = ['id', 'name', 'bio']

class BookSerializer(serializers.ModelSerializer):
    # 嵌套序列化器
    author = AuthorSerializer(read_only=True)
    author_id = serializers.PrimaryKeyRelatedField(
        queryset=Author.objects.all(),
        source='author',
        write_only=True
    )
    
    # 自定义字段
    discount_price = serializers.SerializerMethodField()
    is_new = serializers.BooleanField(read_only=True)
    
    class Meta:
        model = Book
        fields = [
            'id', 'title', 'author', 'author_id', 'price',
            'discount_price', 'is_new', 'published_date'
        ]
        read_only_fields = ['id', 'is_new']
        extra_kwargs = {
            'price': {'min_value': 0, 'required': True},
            'title': {'max_length': 100, 'trim_whitespace': True}
        }
    
    def get_discount_price(self, obj):
        """
        计算折扣价（自定义字段方法）
        """
        return float(obj.price) * 0.8
    
    def validate_title(self, value):
        """
        字段级验证
        """
        if 'python' not in value.lower():
            raise serializers.ValidationError("标题必须包含'python'")
        return value
    
    def validate(self, data):
        """
        对象级验证
        """
        if data['published_date'].year < 2000:
            raise serializers.ValidationError("出版日期必须在2000年后")
        return data
    
    def create(self, validated_data):
        """
        自定义创建逻辑
        """
        # 可以在这里添加额外逻辑
        return super().create(validated_data)

序列化器字段类型

字段类型 说明 示例
CharField 文本字段 CharField(max_length=100, min_length=5)
IntegerField 整数字段 IntegerField(min_value=0, max_value=100)
FloatField 浮点数字段 FloatField()
DecimalField 精确小数字段 DecimalField(max_digits=10, decimal_places=2)
BooleanField 布尔字段 BooleanField(default=False)
DateTimeField 日期时间字段 DateTimeField(read_only=True)
DateField 日期字段 DateField()
EmailField 邮箱字段 EmailField(max_length=100)
URLField URL字段 URLField(max_length=200)
UUIDField UUID字段 UUIDField(format='hex')
FileField 文件字段 FileField(max_length=100)
ImageField 图片字段 ImageField(max_length=100)
JSONField JSON字段 JSONField()
ListField 列表字段 ListField(child=CharField())
DictField 字典字段 DictField(child=CharField())
SerializerMethodField 自定义方法字段 SerializerMethodField()
PrimaryKeyRelatedField 外键关联（ID） PrimaryKeyRelatedField(queryset=Author.objects.all())
StringRelatedField 外键关联（字符串） StringRelatedField()
SlugRelatedField 外键关联（Slug） SlugRelatedField(slug_field='name', queryset=Author.objects.all())
HyperlinkedRelatedField 外键关联（URL） HyperlinkedRelatedField(view_name='author-detail', queryset=Author.objects.all())
HyperlinkedIdentityField 自身URL HyperlinkedIdentityField(view_name='book-detail')
Nested Serializer 嵌套序列化器 author = AuthorSerializer()

序列化器验证

验证示例
class UserSerializer(serializers.Serializer):
    username = serializers.CharField(max_length=100)
    password = serializers.CharField(write_only=True)
    password2 = serializers.CharField(write_only=True)
    
    # 字段级验证：validate_字段名
    def validate_username(self, value):
        if User.objects.filter(username=value).exists():
            raise serializers.ValidationError("用户名已存在")
        return value
    
    def validate_password(self, value):
        if len(value) < 8:
            raise serializers.ValidationError("密码至少8位")
        return value
    
    # 对象级验证：validate
    def validate(self, data):
        if data['password'] != data['password2']:
            raise serializers.ValidationError({"password2": "两次密码不一致"})
        return data

批量序列化

批量操作
# 序列化多个对象
books = Book.objects.all()
serializer = BookSerializer(books, many=True)

# 反序列化多个对象
data = [
    {'title': 'Book 1', 'price': 29.99},
    {'title': 'Book 2', 'price': 39.99}
]
serializer = BookSerializer(data=data, many=True)
if serializer.is_valid():
    serializer.save()

视图 (Views)

视图类型对比

视图类型 适用场景 代码量 灵活性
APIView 完全自定义逻辑 多 最高
GenericAPIView 需要自定义查询/序列化 中 高
Concrete Views 标准CRUD操作 少 中
ViewSets 完整REST接口 最少 中

APIView 基础

views.py
from rest_framework.views import APIView
from rest_framework.response import Response
from rest_framework import status
from .models import Book
from .serializers import BookSerializer

class BookListAPIView(APIView):
    """
    基于APIView的完整示例
    """
    def get(self, request, format=None):
        """
        获取图书列表
        """
        books = Book.objects.all()
        serializer = BookSerializer(books, many=True)
        return Response(serializer.data)
    
    def post(self, request, format=None):
        """
        创建新图书
        """
        serializer = BookSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

class BookDetailAPIView(APIView):
    """
    单资源操作
    """
    def get_object(self, pk):
        try:
            return Book.objects.get(pk=pk)
        except Book.DoesNotExist:
            raise Http404
    
    def get(self, request, pk, format=None):
        book = self.get_object(pk)
        serializer = BookSerializer(book)
        return Response(serializer.data)
    
    def put(self, request, pk, format=None):
        book = self.get_object(pk)
        serializer = BookSerializer(book, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
    
    def delete(self, request, pk, format=None):
        book = self.get_object(pk)
        book.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

GenericAPIView 与 Mixin

views.py
from rest_framework import generics
from rest_framework.mixins import ListModelMixin, CreateModelMixin

class BookListGenericView(generics.GenericAPIView, ListModelMixin, CreateModelMixin):
    """
    使用GenericAPIView和Mixin
    """
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    # 可选：自定义查询
    def get_queryset(self):
        queryset = Book.objects.all()
        is_available = self.request.query_params.get('is_available')
        if is_available is not None:
            queryset = queryset.filter(is_available=is_available)
        return queryset
    
    # 可选：自定义序列化器
    def get_serializer_class(self):
        if self.request.method == 'POST':
            return BookCreateSerializer
        return BookSerializer
    
    def get(self, request, *args, **kwargs):
        return self.list(request, *args, **kwargs)
    
    def post(self, request, *args, **kwargs):
        return self.create(request, *args, **kwargs)

具体通用视图

views.py
from rest_framework import generics

# 列表和创建
class BookListCreateView(generics.ListCreateAPIView):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    # 可选：分页
    pagination_class = StandardResultsSetPagination
    
    # 可选：过滤
    filter_backends = [DjangoFilterBackend, OrderingFilter]
    filterset_fields = ['author', 'is_available']
    ordering_fields = ['price', 'published_date']
    ordering = ['-published_date']

# 详情、更新、删除
class BookDetailView(generics.RetrieveUpdateDestroyAPIView):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    # 可选：部分更新
    def patch(self, request, *args, **kwargs):
        return self.partial_update(request, *args, **kwargs)

视图装饰器

装饰器示例
from rest_framework.decorators import api_view, permission_classes, throttle_classes
from rest_framework.permissions import IsAuthenticated
from rest_framework.throttling import UserRateThrottle

class BurstRateThrottle(UserRateThrottle):
    rate = '10/min'

@api_view(['GET', 'POST'])
@permission_classes([IsAuthenticated])
@throttle_classes([BurstRateThrottle])
def book_list(request):
    """
    函数视图示例
    """
    if request.method == 'GET':
        books = Book.objects.all()
        serializer = BookSerializer(books, many=True)
        return Response(serializer.data)
    
    elif request.method == 'POST':
        serializer = BookSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

视图集 (ViewSets)

基础ViewSet

views.py
from rest_framework import viewsets
from rest_framework.decorators import action
from rest_framework.response import Response

class BookViewSet(viewsets.ViewSet):
    """
    基础ViewSet，需手动实现所有方法
    """
    def list(self, request):
        queryset = Book.objects.all()
        serializer = BookSerializer(queryset, many=True)
        return Response(serializer.data)
    
    def create(self, request):
        serializer = BookSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
    
    def retrieve(self, request, pk=None):
        book = get_object_or_404(Book, pk=pk)
        serializer = BookSerializer(book)
        return Response(serializer.data)
    
    def update(self, request, pk=None):
        book = get_object_or_404(Book, pk=pk)
        serializer = BookSerializer(book, data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)
    
    def destroy(self, request, pk=None):
        book = get_object_or_404(Book, pk=pk)
        book.delete()
        return Response(status=status.HTTP_204_NO_CONTENT)

ModelViewSet（推荐）

views.py
class BookModelViewSet(viewsets.ModelViewSet):
    """
    ModelViewSet自动提供list, create, retrieve, update, destroy
    """
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    # 权限
    permission_classes = [IsAuthenticatedOrReadOnly]
    
    # 过滤和搜索
    filter_backends = [DjangoFilterBackend, SearchFilter, OrderingFilter]
    filterset_fields = ['author', 'is_available']
    search_fields = ['title', 'author__name']
    ordering_fields = ['price', 'published_date']
    
    # 分页
    pagination_class = StandardResultsSetPagination
    
    # 自定义查询集
    def get_queryset(self):
        queryset = Book.objects.select_related('author').all()
        
        # 根据用户过滤
        if self.request.user.is_staff:
            return queryset
        return queryset.filter(is_available=True)
    
    # 自定义序列化器
    def get_serializer_class(self):
        if self.action == 'create':
            return BookCreateSerializer
        elif self.action == 'retrieve':
            return BookDetailSerializer
        return BookSerializer
    
    # 自定义权限
    def get_permissions(self):
        if self.action in ['create', 'update', 'destroy']:
            return [IsAdminUser()]
        return [IsAuthenticatedOrReadOnly()]
    
    # 自定义动作
    @action(detail=True, methods=['post'], permission_classes=[IsAuthenticated])
    def borrow(self, request, pk=None):
        """
        自定义动作：借阅图书
        POST /api/books/1/borrow/
        """
        book = self.get_object()
        # 借阅逻辑...
        return Response({'status': 'book borrowed'})
    
    @action(detail=False, methods=['get'])
    def recommendations(self, request):
        """
        自定义动作：获取推荐
        GET /api/books/recommendations/
        """
        books = Book.objects.filter(rating__gte=4.5)[:5]
        serializer = self.get_serializer(books, many=True)
        return Response(serializer.data)
    
    @action(detail=False, methods=['post'], url_path='bulk-create')
    def bulk_create(self, request):
        """
        批量创建
        POST /api/books/bulk-create/
        """
        serializer = self.get_serializer(data=request.data, many=True)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

ReadOnlyModelViewSet

views.py
class BookReadOnlyViewSet(viewsets.ReadOnlyModelViewSet):
    """
    只读视图集，仅提供list和retrieve
    """
    queryset = Book.objects.filter(is_available=True)
    serializer_class = BookSerializer
    pagination_class = None  # 禁用分页

ViewSet路由

urls.py
from rest_framework.routers import DefaultRouter, SimpleRouter

# DefaultRouter包含API根视图和默认路由
router = DefaultRouter()
router.register(r'books', BookModelViewSet, basename='book')
router.register(r'authors', AuthorViewSet)

# SimpleRouter不包含API根视图
# router = SimpleRouter()

urlpatterns = [
    path('', include(router.urls)),
]

# 生成的URL：
# GET/POST      /api/books/          -> list/create
# GET/PUT/DELETE /api/books/1/       -> retrieve/update/destroy
# POST          /api/books/1/borrow/ -> borrow action
# GET           /api/books/recommendations/ -> recommendations action

路由 (Routers)

路由器类型

路由器 API根视图 尾随斜杠 适用场景
DefaultRouter 有 有 标准REST API
SimpleRouter 无 有 不需要API根

自定义路由

urls.py
from rest_framework.routers import Route, DynamicRoute, SimpleRouter

class CustomRouter(SimpleRouter):
    """
    自定义路由器
    """
    routes = [
        # 列表路由
        Route(
            url=r'^{prefix}$',
            mapping={'get': 'list', 'post': 'create'},
            name='{basename}-list',
            detail=False,
            initkwargs={'suffix': 'List'}
        ),
        # 详情路由
        Route(
            url=r'^{prefix}/{lookup}$',
            mapping={'get': 'retrieve', 'put': 'update', 'delete': 'destroy'},
            name='{basename}-detail',
            detail=True,
            initkwargs={'suffix': 'Instance'}
        ),
        # 自定义动态路由
        DynamicRoute(
            url=r'^{prefix}/{lookup}/{url_path}$',
            name='{basename}-{url_name}',
            detail=True,
            initkwargs={}
        )
    ]

router = CustomRouter()
router.register(r'books', BookViewSet, basename='book')

解析器 (Parsers)

内置解析器

解析器 Content-Type 说明
JSONParser application/json 解析JSON数据
FormParser application/x-www-form-urlencoded 解析表单数据
MultiPartParser multipart/form-data 解析文件上传
FileUploadParser */* 原生文件上传

配置解析器

settings.py
REST_FRAMEWORK = {
    'DEFAULT_PARSER_CLASSES': [
        'rest_framework.parsers.JSONParser',
        'rest_framework.parsers.FormParser',
        'rest_framework.parsers.MultiPartParser',
    ]
}

视图级别配置

views.py
from rest_framework.parsers import JSONParser, FormParser

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    parser_classes = [JSONParser, FormParser]  # 仅该视图使用

渲染器 (Renderers)

内置渲染器

渲染器 Media Type 说明
JSONRenderer application/json JSON格式（默认）
BrowsableAPIRenderer text/html 可浏览API界面
TemplateHTMLRenderer text/html HTML模板渲染
StaticHTMLRenderer text/html 静态HTML
AdminRenderer text/html Admin风格界面

配置渲染器

settings.py
REST_FRAMEWORK = {
    'DEFAULT_RENDERER_CLASSES': [
        'rest_framework.renderers.JSONRenderer',
        'rest_framework.renderers.BrowsableAPIRenderer',
    ]
}

条件渲染（生产环境禁用Browsable API）

settings.py
if DEBUG:
    REST_FRAMEWORK['DEFAULT_RENDERER_CLASSES'].append(
        'rest_framework.renderers.BrowsableAPIRenderer'
    )

认证 (Authentication)

内置认证类

认证类 说明 适用场景
SessionAuthentication Django会话认证 Web浏览器
BasicAuthentication HTTP Basic Auth 测试、内部API
TokenAuthentication Token认证 移动端、第三方
RemoteUserAuthentication 远程用户认证 SSO集成

Token认证配置

settings.py
INSTALLED_APPS = [
    ...
    'rest_framework.authtoken',
]

REST_FRAMEWORK = {
    'DEFAULT_AUTHENTICATION_CLASSES': [
        'rest_framework.authentication.TokenAuthentication',
        'rest_framework.authentication.SessionAuthentication',
    ]
}
urls.py
from rest_framework.authtoken.views import obtain_auth_token

urlpatterns = [
    ...
    path('api-token-auth/', obtain_auth_token),  # 获取token
]

自定义JWT认证（推荐）

authentication.py
import jwt
from django.conf import settings
from django.contrib.auth import get_user_model
from rest_framework import authentication, exceptions
from datetime import datetime, timedelta

User = get_user_model()

class JWTAuthentication(authentication.BaseAuthentication):
    """
    自定义JWT认证
    """
    def authenticate(self, request):
        auth_header = authentication.get_authorization_header(request).split()
        
        if not auth_header or auth_header[0].lower() != b'bearer':
            return None
        
        if len(auth_header) == 1:
            raise exceptions.AuthenticationFailed('Invalid token header')
        
        try:
            token = auth_header[1].decode('utf-8')
            payload = jwt.decode(token, settings.SECRET_KEY, algorithms=['HS256'])
        except jwt.ExpiredSignatureError:
            raise exceptions.AuthenticationFailed('Token已过期')
        except jwt.InvalidTokenError:
            raise exceptions.AuthenticationFailed('无效的Token')
        
        try:
            user = User.objects.get(id=payload['user_id'])
        except User.DoesNotExist:
            raise exceptions.AuthenticationFailed('用户不存在')
        
        if not user.is_active:
            raise exceptions.AuthenticationFailed('用户已被禁用')
        
        return (user, token)

def generate_token(user):
    """
    生成JWT Token
    """
    payload = {
        'user_id': user.id,
        'exp': datetime.utcnow() + timedelta(days=7),
        'iat': datetime.utcnow(),
    }
    return jwt.encode(payload, settings.SECRET_KEY, algorithm='HS256')

多平台统一认证（小程序+后台）

authentication.py
class UnifiedAuthentication(authentication.BaseAuthentication):
    """
    支持小程序和后台统一认证
    """
    def authenticate(self, request):
        auth_header = authentication.get_authorization_header(request).split()
        
        if not auth_header:
            return None
        
        token = auth_header[0].decode('utf-8')
        
        # 根据token类型分发到不同认证后端
        if token.startswith('mini_'):
            return self._authenticate_miniapp(token)
        else:
            return self._authenticate_admin(token)
    
    def _authenticate_miniapp(self, token):
        # 小程序认证逻辑
        pass
    
    def _authenticate_admin(self, token):
        # 后台认证逻辑
        pass

权限 (Permissions)

内置权限类

权限类 说明
AllowAny 允许所有访问
IsAuthenticated 仅认证用户
IsAdminUser 仅管理员（is_staff=True）
IsAuthenticatedOrReadOnly 认证用户可写，匿名只读
DjangoModelPermissions 基于Django模型权限
DjangoObjectPermissions 基于Django对象权限

自定义权限

permissions.py
from rest_framework import permissions

class IsOwnerOrReadOnly(permissions.BasePermission):
    """
    对象级权限：仅所有者可以编辑
    """
    def has_object_permission(self, request, view, obj):
        # 安全方法（GET, HEAD, OPTIONS）允许所有
        if request.method in permissions.SAFE_METHODS:
            return True
        
        # 写权限仅所有者有
        return obj.owner == request.user

class IsAdminOrReadOnly(permissions.BasePermission):
    """
    管理员可写，其他只读
    """
    def has_permission(self, request, view):
        if request.method in permissions.SAFE_METHODS:
            return True
        return request.user and request.user.is_staff

class HasRolePermission(permissions.BasePermission):
    """
    基于角色的权限
    """
    required_roles = []
    
    def has_permission(self, request, view):
        if not request.user or not request.user.is_authenticated:
            return False
        
        # 超级管理员拥有所有权限
        if request.user.is_superuser:
            return True
        
        # 检查用户角色
        user_roles = getattr(request.user, 'roles', [])
        return any(role in user_roles for role in self.required_roles)

# 使用示例
class BookViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated, IsOwnerOrReadOnly]

动态权限

views.py
class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    def get_permissions(self):
        """
        根据动作返回不同权限
        """
        if self.action == 'list':
            return [AllowAny()]
        elif self.action == 'create':
            return [IsAuthenticated()]
        elif self.action in ['update', 'partial_update', 'destroy']:
            return [IsAuthenticated(), IsOwnerOrReadOnly()]
        return [IsAuthenticated()]

限流 (Throttling)

内置限流类

限流类 说明
AnonRateThrottle 匿名用户限流
UserRateThrottle 认证用户限流
ScopedRateThrottle 基于作用域的限流

配置限流

settings.py
REST_FRAMEWORK = {
    'DEFAULT_THROTTLE_CLASSES': [
        'rest_framework.throttling.AnonRateThrottle',
        'rest_framework.throttling.UserRateThrottle',
        'rest_framework.throttling.ScopedRateThrottle',
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/day',
        'user': '1000/day',
        'burst': '60/min',
        'sustained': '1000/day',
        'book_list': '30/min',  # 自定义作用域
    }
}

自定义限流

throttling.py
from rest_framework.throttling import SimpleRateThrottle

class BurstRateThrottle(SimpleRateThrottle):
    """
    突发流量限流
    """
    scope = 'burst'
    
    def get_cache_key(self, request, view):
        # 根据用户ID或IP限流
        if request.user.is_authenticated:
            ident = request.user.pk
        else:
            ident = self.get_ident(request)
        return self.cache_format % {'scope': self.scope, 'ident': ident}

class APIKeyRateThrottle(SimpleRateThrottle):
    """
    基于API Key的限流
    """
    scope = 'api_key'
    
    def get_cache_key(self, request, view):
        api_key = request.headers.get('X-API-Key')
        if api_key:
            return f"throttle_{self.scope}_{api_key}"
        return None

# 视图使用
class BookViewSet(viewsets.ModelViewSet):
    throttle_classes = [BurstRateThrottle]

过滤 (Filtering)

内置过滤

views.py
from rest_framework import filters

class BookListView(generics.ListAPIView):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    filter_backends = [filters.SearchFilter, filters.OrderingFilter]
    search_fields = ['title', 'author__name', 'description']
    ordering_fields = ['price', 'published_date', 'rating']
    ordering = ['-published_date']  # 默认排序

DjangoFilterBackend（推荐）

安装
pip install django-filter
settings.py
INSTALLED_APPS = [
    'django_filters',
]

REST_FRAMEWORK = {
    'DEFAULT_FILTER_BACKENDS': [
        'django_filters.rest_framework.DjangoFilterBackend',
    ]
}
filters.py
import django_filters
from .models import Book

class BookFilter(django_filters.FilterSet):
    """
    自定义过滤器
    """
    min_price = django_filters.NumberFilter(field_name='price', lookup_expr='gte')
    max_price = django_filters.NumberFilter(field_name='price', lookup_expr='lte')
    title_contains = django_filters.CharFilter(field_name='title', lookup_expr='icontains')
    published_after = django_filters.DateFilter(field_name='published_date', lookup_expr='gte')
    published_before = django_filters.DateFilter(field_name='published_date', lookup_expr='lte')
    author_name = django_filters.CharFilter(field_name='author__name', lookup_expr='iexact')
    tags = django_filters.CharFilter(method='filter_tags')
    
    class Meta:
        model = Book
        fields = ['is_available', 'category', 'min_price', 'max_price']
    
    def filter_tags(self, queryset, name, value):
        tags = value.split(',')
        return queryset.filter(tags__name__in=tags)
views.py
from django_filters.rest_framework import DjangoFilterBackend

class BookViewSet(viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_class = BookFilter
    search_fields = ['title', 'description']
    ordering_fields = ['price', 'published_date']

自定义过滤

views.py
class BookViewSet(viewsets.ModelViewSet):
    serializer_class = BookSerializer
    
    def get_queryset(self):
        queryset = Book.objects.all()
        
        # 根据查询参数过滤
        is_available = self.request.query_params.get('is_available')
        if is_available is not None:
            queryset = queryset.filter(is_available=is_available.lower() == 'true')
        
        min_price = self.request.query_params.get('min_price')
        if min_price:
            queryset = queryset.filter(price__gte=float(min_price))
        
        # 根据用户过滤
        if not self.request.user.is_staff:
            queryset = queryset.filter(is_public=True)
        
        return queryset

版本控制 (Versioning)

版本控制方案

方案 URL示例 说明
URL路径 /api/v1/books/ 推荐，最清晰
查询参数 /api/books/?version=1.0 简单但不够RESTful
请求头 Accept: application/json; version=1.0 符合REST规范
主机名 v1.api.example.com 需要DNS配置

URL路径版本配置

settings.py
REST_FRAMEWORK = {
    'DEFAULT_VERSIONING_CLASS': 'rest_framework.versioning.URLPathVersioning',
    'DEFAULT_VERSION': 'v1',
    'ALLOWED_VERSIONS': ['v1', 'v2'],
    'VERSION_PARAM': 'version',
}
urls.py
urlpatterns = [
    path('api//', include('myapp.urls')),
]

视图中使用版本

views.py
class BookViewSet(viewsets.ModelViewSet):
    def get_serializer_class(self):
        if self.request.version == 'v1':
            return BookSerializerV1
        return BookSerializerV2
    
    def get_queryset(self):
        if self.request.version == 'v1':
            return Book.objects.filter(is_legacy=True)
        return Book.objects.all()

缓存 (Caching)

视图缓存

views.py
from django.views.decorators.cache import cache_page
from django.utils.decorators import method_decorator

class BookListView(APIView):
    @method_decorator(cache_page(60 * 15))  # 缓存15分钟
    def get(self, request):
        books = Book.objects.all()
        serializer = BookSerializer(books, many=True)
        return Response(serializer.data)

DRF缓存扩展

安装
pip install drf-extensions
views.py
from rest_framework_extensions.cache.decorators import cache_response
from rest_framework_extensions.cache.mixins import CacheResponseMixin

class BookViewSet(CacheResponseMixin, viewsets.ModelViewSet):
    queryset = Book.objects.all()
    serializer_class = BookSerializer
    
    # 缓存时间
    cache_response_timeout = 60 * 15
    
    # 自定义缓存键
    def get_cache_key(self, request, *args, **kwargs):
        return f"book_list_{request.query_params.get('category', 'all')}"

可浏览API (Browsable API)

自定义样式

settings.py
REST_FRAMEWORK = {
    'DEFAULT_RENDERER_CLASSES': [
        'rest_framework.renderers.JSONRenderer',
        'rest_framework.renderers.BrowsableAPIRenderer',
    ],
    'DEFAULT_METADATA_CLASS': 'rest_framework.metadata.SimpleMetadata',
}

自定义模板

templates/rest_framework/api.html
{% extends "rest_framework/base.html" %}

{% block branding %}
    
        My Project API
    
{% endblock %}

{% block style %}
    {{ block.super }}
    
{% endblock %}

Admin集成

admin.py
from django.contrib import admin
from .models import Book

@admin.register(Book)
class BookAdmin(admin.ModelAdmin):
    list_display = ['title', 'author', 'price', 'is_available']
    list_filter = ['is_available', 'published_date']
    search_fields = ['title', 'author__name']
    date_hierarchy = 'published_date'

异常处理

自定义异常

exceptions.py
from rest_framework.exceptions import APIException
from rest_framework import status

class ServiceUnavailable(APIException):
    status_code = status.HTTP_503_SERVICE_UNAVAILABLE
    default_detail = '服务暂时不可用'
    default_code = 'service_unavailable'

class InsufficientFunds(APIException):
    status_code = status.HTTP_402_PAYMENT_REQUIRED
    default_detail = '余额不足'
    default_code = 'insufficient_funds'

# 使用
raise ServiceUnavailable()

全局异常处理

utils.py
from rest_framework.views import exception_handler
from rest_framework.response import Response

def custom_exception_handler(exc, context):
    """
    自定义异常处理
    """
    response = exception_handler(exc, context)
    
    if response is not None:
        # 添加自定义数据
        response.data['status_code'] = response.status_code
        
        # 统一错误格式
        if 'detail' in response.data:
            response.data['message'] = response.data.pop('detail')
    
    return response
settings.py
REST_FRAMEWORK = {
    'EXCEPTION_HANDLER': 'myapp.utils.custom_exception_handler',
}

测试工具

API测试

tests.py
from rest_framework.test import APITestCase, APIClient
from rest_framework import status
from django.urls import reverse

class BookAPITests(APITestCase):
    def setUp(self):
        self.client = APIClient()
        self.user = User.objects.create_user(username='test', password='123456')
        self.book = Book.objects.create(title='Test Book', price=29.99)
    
    def test_list_books(self):
        """
        测试获取图书列表
        """
        url = reverse('book-list')
        response = self.client.get(url)
        self.assertEqual(response.status_code, status.HTTP_200_OK)
        self.assertEqual(len(response.data['results']), 1)
    
    def test_create_book(self):
        """
        测试创建图书（需认证）
        """
        self.client.force_authenticate(user=self.user)
        url = reverse('book-list')
        data = {
            'title': 'New Book',
            'price': 39.99
        }
        response = self.client.post(url, data, format='json')
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)
        self.assertEqual(Book.objects.count(), 2)
    
    def test_unauthorized_access(self):
        """
        测试未授权访问
        """
        url = reverse('book-list')
        response = self.client.post(url, {}, format='json')
        self.assertEqual(response.status_code, status.HTTP_401_UNAUTHORIZED)

Schema与文档

自动生成API文档

安装
pip install drf-spectacular  # OpenAPI 3
settings.py
INSTALLED_APPS = [
    'drf_spectacular',
]

REST_FRAMEWORK = {
    'DEFAULT_SCHEMA_CLASS': 'drf_spectacular.openapi.AutoSchema',
}

SPECTACULAR_SETTINGS = {
    'TITLE': 'My Project API',
    'DESCRIPTION': 'API documentation',
    'VERSION': '1.0.0',
}
urls.py
from drf_spectacular.views import SpectacularAPIView, SpectacularRedocView, SpectacularSwaggerView

urlpatterns = [
    path('api/schema/', SpectacularAPIView.as_view(), name='schema'),
    path('api/docs/swagger/', SpectacularSwaggerView.as_view(url_name='schema'), name='swagger-ui'),
    path('api/docs/redoc/', SpectacularRedocView.as_view(url_name='schema'), name='redoc'),
]

性能优化

查询优化

views.py
class BookViewSet(viewsets.ModelViewSet):
    # 使用select_related减少查询
    queryset = Book.objects.select_related('author', 'publisher').all()
    
    # 使用prefetch_related处理多对多
    queryset = Book.objects.prefetch_related('tags', 'reviews').all()
    
    # 仅获取需要的字段
    queryset = Book.objects.only('title', 'price', 'author__name')

序列化器优化

serializers.py
class BookSerializer(serializers.ModelSerializer):
    # 使用StringRelatedField减少序列化开销
    author = serializers.StringRelatedField()
    
    # 使用ReadOnlyField
    is_new = serializers.ReadOnlyField()
    
    class Meta:
        model = Book
        fields = ['id', 'title', 'author', 'is_new']
        # 排除大字段
        exclude = ['description', 'content']

部署建议

settings.py (production)
# 生产环境配置
DEBUG = False

REST_FRAMEWORK = {
    'DEFAULT_RENDERER_CLASSES': [
        'rest_framework.renderers.JSONRenderer',  # 仅JSON
    ],
    'DEFAULT_THROTTLE_RATES': {
        'anon': '100/hour',
        'user': '1000/hour',
    },
}

# 启用gzip压缩
MIDDLEWARE = [
    'django.middleware.gzip.GZipMiddleware',
    # ...
]

# 数据库连接池
DATABASES = {
    'default': {
        'ENGINE': 'django.db.backends.postgresql',
        'CONN_MAX_AGE': 600,  # 连接持久化
    }
}

配置项 (Settings)

配置项 默认值 说明
DEFAULT_AUTHENTICATION_CLASSES SessionAuthentication 默认认证类
DEFAULT_PERMISSION_CLASSES AllowAny 默认权限类
DEFAULT_THROTTLE_CLASSES [] 默认限流类
DEFAULT_PARSER_CLASSES JSONParser, FormParser, MultiPartParser 默认解析器
DEFAULT_RENDERER_CLASSES JSONRenderer, BrowsableAPIRenderer 默认渲染器
DEFAULT_PAGINATION_CLASS None 默认分页类
PAGE_SIZE None 默认分页大小
DEFAULT_FILTER_BACKENDS [] 默认过滤后端
DEFAULT_VERSIONING_CLASS None 默认版本控制类
EXCEPTION_HANDLER exception_handler 异常处理函数
TEST_REQUEST_RENDERER_CLASSES JSONRenderer, MultiPartRenderer 测试请求渲染器

HTTP状态码

rest_framework.status
HTTP_200_OK = 200
HTTP_201_CREATED = 201
HTTP_204_NO_CONTENT = 204
HTTP_400_BAD_REQUEST = 400
HTTP_401_UNAUTHORIZED = 401
HTTP_403_FORBIDDEN = 403
HTTP_404_NOT_FOUND = 404
HTTP_405_METHOD_NOT_ALLOWED = 405
HTTP_429_TOO_MANY_REQUESTS = 429
HTTP_500_INTERNAL_SERVER_ERROR = 500

装饰器

装饰器 说明 示例
@api_view 函数视图 @api_view(['GET', 'POST'])
@permission_classes 权限控制 @permission_classes([IsAuthenticated])
@authentication_classes 认证控制 @authentication_classes([TokenAuthentication])
@throttle_classes 限流控制 @throttle_classes([UserRateThrottle])
@parser_classes 解析器控制 @parser_classes([JSONParser])
@renderer_classes 渲染器控制 @renderer_classes([JSONRenderer])
@schema Schema控制 @schema(None) # 排除文档

中间件

middleware.py
class APIVersionMiddleware:
    """
    API版本中间件
    """
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        # 从请求头获取版本
        version = request.META.get('HTTP_API_VERSION', 'v1')
        request.version = version
        
        response = self.get_response(request)
        response['API-Version'] = version
        return response

class RequestLogMiddleware:
    """
    请求日志中间件
    """
    def __init__(self, get_response):
        self.get_response = get_response
    
    def __call__(self, request):
        import time
        start_time = time.time()
        
        response = self.get_response(request)
        
        duration = time.time() - start_time
        print(f"[{request.method}] {request.path} - {response.status_code} ({duration:.2f}s)")
        
        return response

Django REST Framework 完整文档

生成时间: 2026-03-10 | 版本: 3.14+