Django REST Framework实战：从零构建企业级API服务-程序员充电站

Django REST Framework实战：从零构建企业级API服务

【免费下载链接】Python-100-DaysPython - 100天从新手到大师项目地址: https://gitcode.com/GitHub_Trending/py/Python-100-Days

在当今前后端分离的开发模式中，API已成为连接客户端与服务端的核心桥梁。然而，许多开发者在构建API时常常陷入困惑：如何设计既规范又高效的接口？如何确保API的安全性？如何处理复杂的业务逻辑？本文将带你从实际问题出发，通过Django REST Framework（DRF）构建一套完整的RESTful API解决方案。

为什么选择DRF：解决传统开发痛点

在传统Web开发中，前后端耦合严重，每次需求变更都需要双方协调修改。RESTful架构的出现，让前后端能够独立开发、独立部署。DRF作为Django生态中最成熟的API框架，提供了以下核心优势：

开箱即用的API视图：基于类的视图系统让CRUD操作变得异常简单
强大的序列化器：自动处理数据验证和转换，支持复杂嵌套关系
丰富的认证授权机制：从基础认证到JWT，满足不同安全需求
可扩展的过滤和分页：轻松处理大数据集的查询和展示
自动生成的API文档：减少文档编写和维护成本

项目实战：构建图书管理系统API

让我们通过一个实际的图书管理系统案例，逐步掌握DRF的核心技术。假设我们需要为图书馆开发一套API，支持图书的增删改查、借阅记录管理等功能。

第一步：环境搭建与基础配置

首先创建Django项目并安装DRF：

# 创建项目目录 mkdir library_api && cd library_api # 创建虚拟环境 python -m venv venv source venv/bin/activate # Linux/Mac # 或 venv\Scripts\activate # Windows # 安装依赖 pip install django djangorestframework django-filter # 创建Django项目 django-admin startproject library . django-admin startapp books

配置settings.py中的DRF：

# settings.py INSTALLED_APPS = [ 'django.contrib.admin', 'django.contrib.auth', 'django.contrib.contenttypes', 'django.contrib.sessions', 'django.contrib.messages', 'django.contrib.staticfiles', 'rest_framework', 'books', ] REST_FRAMEWORK = { 'DEFAULT_PERMISSION_CLASSES': [ 'rest_framework.permissions.IsAuthenticatedOrReadOnly', ], 'DEFAULT_AUTHENTICATION_CLASSES': [ 'rest_framework.authentication.SessionAuthentication', 'rest_framework.authentication.BasicAuthentication', ], 'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination', 'PAGE_SIZE': 20 }

第二步：设计数据模型

良好的数据库设计是API稳定性的基础。我们创建图书和借阅记录两个主要模型：

# books/models.py from django.db import models from django.contrib.auth.models import User class Book(models.Model): """图书模型""" CATEGORY_CHOICES = [ ('fiction', '小说'), ('science', '科技'), ('history', '历史'), ('education', '教育'), ('other', '其他'), ] isbn = models.CharField('ISBN号', max_length=13, unique=True) title = models.CharField('书名', max_length=200) author = models.CharField('作者', max_length=100) publisher = models.CharField('出版社', max_length=100) publish_date = models.DateField('出版日期') category = models.CharField('分类', max_length=20, choices=CATEGORY_CHOICES) price = models.DecimalField('价格', max_digits=8, decimal_places=2) stock = models.IntegerField('库存数量', default=1) available = models.BooleanField('可借阅', default=True) created_at = models.DateTimeField('创建时间', auto_now_add=True) updated_at = models.DateTimeField('更新时间', auto_now=True) class Meta: ordering = ['-created_at'] verbose_name = '图书' verbose_name_plural = '图书' def __str__(self): return f"{self.title} - {self.author}" class BorrowRecord(models.Model): """借阅记录模型""" STATUS_CHOICES = [ ('borrowed', '借阅中'), ('returned', '已归还'), ('overdue', '已逾期'), ] book = models.ForeignKey(Book, on_delete=models.CASCADE, related_name='borrow_records') user = models.ForeignKey(User, on_delete=models.CASCADE, related_name='borrow_records') borrow_date = models.DateField('借阅日期', auto_now_add=True) due_date = models.DateField('应还日期') return_date = models.DateField('归还日期', null=True, blank=True) status = models.CharField('状态', max_length=20, choices=STATUS_CHOICES, default='borrowed') created_at = models.DateTimeField('创建时间', auto_now_add=True) class Meta: ordering = ['-borrow_date'] verbose_name = '借阅记录' verbose_name_plural = '借阅记录' def __str__(self): return f"{self.user.username}借阅《{self.book.title}》"

第三步：创建序列化器

序列化器是DRF的核心组件，负责数据验证和转换。我们为图书模型创建两个序列化器：一个用于列表展示，一个用于详细展示：

# books/serializers.py from rest_framework import serializers from .models import Book, BorrowRecord from django.contrib.auth.models import User class BookListSerializer(serializers.ModelSerializer): """图书列表序列化器""" category_display = serializers.CharField(source='get_category_display', read_only=True) class Meta: model = Book fields = ['id', 'isbn', 'title', 'author', 'category', 'category_display', 'price', 'available', 'stock'] def validate_isbn(self, value): """验证ISBN号格式""" if len(value) not in [10, 13]: raise serializers.ValidationError("ISBN号必须是10位或13位") return value def validate_stock(self, value): """验证库存数量""" if value < 0: raise serializers.ValidationError("库存数量不能为负数") return value class BookDetailSerializer(serializers.ModelSerializer): """图书详情序列化器""" category_display = serializers.CharField(source='get_category_display', read_only=True) borrow_count = serializers.SerializerMethodField() class Meta: model = Book fields = '__all__' def get_borrow_count(self, obj): """获取借阅次数""" return obj.borrow_records.count() class UserSerializer(serializers.ModelSerializer): """用户序列化器""" class Meta: model = User fields = ['id', 'username', 'email', 'first_name', 'last_name'] class BorrowRecordSerializer(serializers.ModelSerializer): """借阅记录序列化器""" book_title = serializers.CharField(source='book.title', read_only=True) user_name = serializers.CharField(source='user.username', read_only=True) class Meta: model = BorrowRecord fields = ['id', 'book', 'book_title', 'user', 'user_name', 'borrow_date', 'due_date', 'return_date', 'status'] read_only_fields = ['borrow_date', 'status'] def validate(self, data): """验证借阅逻辑""" book = data.get('book') user = data.get('user') # 检查图书是否可借 if not book.available: raise serializers.ValidationError("该图书暂不可借") # 检查用户是否已借阅该图书 if BorrowRecord.objects.filter(book=book, user=user, status='borrowed').exists(): raise serializers.ValidationError("您已经借阅了该图书") # 检查用户借阅数量限制（假设最多借5本） borrowed_count = BorrowRecord.objects.filter(user=user, status='borrowed').count() if borrowed_count >= 5: raise serializers.ValidationError("您已达到最大借阅数量") return data

第四步：实现视图和路由

DRF提供了多种视图实现方式，我们选择最灵活的视图集（ViewSet）：

# books/views.py from rest_framework import viewsets, filters, status from rest_framework.decorators import action from rest_framework.response import Response from rest_framework.permissions import IsAuthenticated, IsAdminUser from django_filters.rest_framework import DjangoFilterBackend from .models import Book, BorrowRecord from .serializers import BookListSerializer, BookDetailSerializer, BorrowRecordSerializer class BookViewSet(viewsets.ModelViewSet): """图书视图集""" queryset = Book.objects.all() filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter] filterset_fields = ['category', 'available'] search_fields = ['title', 'author', 'isbn'] ordering_fields = ['title', 'author', 'price', 'created_at'] def get_serializer_class(self): """根据动作选择不同的序列化器""" if self.action == 'list': return BookListSerializer return BookDetailSerializer def get_permissions(self): """根据动作设置权限""" if self.action in ['create', 'update', 'partial_update', 'destroy']: permission_classes = [IsAdminUser] else: permission_classes = [IsAuthenticated] return [permission() for permission in permission_classes] @action(detail=True, methods=['post']) def borrow(self, request, pk=None): """借阅图书""" book = self.get_object() user = request.user # 创建借阅记录 borrow_record = BorrowRecord.objects.create( book=book, user=user, due_date=request.data.get('due_date') ) # 更新图书库存 if book.stock > 0: book.stock -= 1 if book.stock == 0: book.available = False book.save() serializer = BorrowRecordSerializer(borrow_record) return Response(serializer.data, status=status.HTTP_201_CREATED) @action(detail=True, methods=['post']) def return_book(self, request, pk=None): """归还图书""" book = self.get_object() user = request.user # 查找未归还的借阅记录 borrow_record = BorrowRecord.objects.filter( book=book, user=user, status='borrowed' ).first() if not borrow_record: return Response( {'error': '没有找到对应的借阅记录'}, status=status.HTTP_404_NOT_FOUND ) # 更新借阅记录 borrow_record.status = 'returned' borrow_record.return_date = request.data.get('return_date') borrow_record.save() # 更新图书库存 book.stock += 1 book.available = True book.save() serializer = BorrowRecordSerializer(borrow_record) return Response(serializer.data) class BorrowRecordViewSet(viewsets.ReadOnlyModelViewSet): """借阅记录视图集（只读）""" serializer_class = BorrowRecordSerializer permission_classes = [IsAuthenticated] filter_backends = [DjangoFilterBackend, filters.OrderingFilter] filterset_fields = ['status'] ordering_fields = ['borrow_date', 'due_date'] def get_queryset(self): """用户只能查看自己的借阅记录""" return BorrowRecord.objects.filter(user=self.request.user)

配置路由：

# books/urls.py from django.urls import path, include from rest_framework.routers import DefaultRouter from .views import BookViewSet, BorrowRecordViewSet router = DefaultRouter() router.register(r'books', BookViewSet, basename='book') router.register(r'borrow-records', BorrowRecordViewSet, basename='borrow-record') urlpatterns = [ path('api/', include(router.urls)), ] # 项目主路由 # library/urls.py from django.contrib import admin from django.urls import path, include urlpatterns = [ path('admin/', admin.site.urls), path('', include('books.urls')), ]

第五步：API认证与权限控制

安全是API设计的重中之重。DRF提供了多种认证方式，我们选择JWT作为认证方案：

# 安装JWT支持 pip install djangorestframework-simplejwt # settings.py中添加配置 REST_FRAMEWORK = { 'DEFAULT_AUTHENTICATION_CLASSES': [ 'rest_framework_simplejwt.authentication.JWTAuthentication', ], 'DEFAULT_PERMISSION_CLASSES': [ 'rest_framework.permissions.IsAuthenticated', ], } # urls.py中添加JWT路由 from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView urlpatterns = [ path('api/token/', TokenObtainPairView.as_view(), name='token_obtain_pair'), path('api/token/refresh/', TokenRefreshView.as_view(), name='token_refresh'), # ... 其他路由 ]

JWT（JSON Web Token）是目前最流行的API认证方案之一。它由三部分组成：头部（Header）、载荷（Payload）和签名（Signature）。JWT的工作流程如下：

客户端使用用户名密码登录，服务器验证后生成JWT
服务器将JWT返回给客户端
客户端在后续请求中携带JWT（通常放在Authorization头中）
服务器验证JWT的签名，确认用户身份

第六步：API文档与测试

DRF自动生成可交互的API文档，让测试变得简单：

# 安装API文档支持 pip install drf-yasg # 配置文档生成 # urls.py from rest_framework import permissions from drf_yasg.views import get_schema_view from drf_yasg import openapi schema_view = get_schema_view( openapi.Info( title="图书管理系统API", default_version='v1', description="图书馆图书管理系统的RESTful API文档", contact=openapi.Contact(email="contact@library.com"), ), public=True, permission_classes=[permissions.AllowAny], ) urlpatterns = [ path('swagger/', schema_view.with_ui('swagger', cache_timeout=0), name='schema-swagger-ui'), path('redoc/', schema_view.with_ui('redoc', cache_timeout=0), name='schema-redoc'), # ... 其他路由 ]

访问/swagger/或/redoc/即可看到完整的API文档。DRF的自动文档界面让API测试变得直观：

第七步：性能优化与高级特性

1. 缓存策略

对于频繁查询但更新不频繁的数据，使用缓存可以显著提升性能：

# settings.py CACHES = { 'default': { 'BACKEND': 'django.core.cache.backends.redis.RedisCache', 'LOCATION': 'redis://127.0.0.1:6379/1', } } # views.py中使用缓存 from django.utils.decorators import method_decorator from django.views.decorators.cache import cache_page from django.views.decorators.vary import vary_on_headers class BookViewSet(viewsets.ModelViewSet): # ... @method_decorator(cache_page(60 * 15)) # 缓存15分钟 @method_decorator(vary_on_headers("Authorization", "Accept-Language")) def list(self, request, *args, **kwargs): return super().list(request, *args, **kwargs)

2. 异步任务处理

对于耗时操作（如发送邮件、生成报表），使用Celery进行异步处理：

# tasks.py from celery import shared_task from django.core.mail import send_mail @shared_task def send_borrow_notification(user_email, book_title, due_date): """异步发送借阅通知邮件""" subject = f"借阅提醒：《{book_title}》即将到期" message = f"您借阅的《{book_title}》将于{due_date}到期，请及时归还。" send_mail(subject, message, 'noreply@library.com', [user_email]) # views.py中调用异步任务 @action(detail=True, methods=['post']) def borrow(self, request, pk=None): # ... 借阅逻辑 # 异步发送通知邮件 send_borrow_notification.delay( user.email, book.title, borrow_record.due_date.strftime('%Y-%m-%d') )

3. 限流与防护

防止API被恶意请求：

# settings.py REST_FRAMEWORK = { 'DEFAULT_THROTTLE_CLASSES': [ 'rest_framework.throttling.AnonRateThrottle', 'rest_framework.throttling.UserRateThrottle' ], 'DEFAULT_THROTTLE_RATES': { 'anon': '100/day', # 匿名用户每天100次 'user': '1000/day', # 认证用户每天1000次 } }

常见问题与解决方案

问题1：N+1查询问题

当序列化器包含外键关联时，可能会出现N+1查询问题。解决方案是使用select_related和prefetch_related：

class BookViewSet(viewsets.ModelViewSet): def get_queryset(self): # 优化查询，减少数据库访问次数 return Book.objects.select_related('publisher').prefetch_related('authors').all()

问题2：API版本管理

随着业务发展，API需要升级但又要保持向后兼容：

# urls.py from django.urls import path, include from rest_framework.routers import DefaultRouter router_v1 = DefaultRouter() router_v1.register(r'books', BookViewSetV1, basename='book-v1') router_v2 = DefaultRouter() router_v2.register(r'books', BookViewSetV2, basename='book-v2') urlpatterns = [ path('api/v1/', include(router_v1.urls)), path('api/v2/', include(router_v2.urls)), ]

问题3：复杂查询参数处理

对于复杂的查询需求，可以使用django-filter的高级功能：

# 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') author = django_filters.CharFilter(field_name="author", lookup_expr='icontains') published_after = django_filters.DateFilter(field_name="publish_date", lookup_expr='gte') class Meta: model = Book fields = ['category', 'available']