第19章 API开发

学习目标

完成本章学习后，读者应能够：

理解REST架构：掌握RESTful API的设计原则、约束条件与最佳实践
精通DRF框架：运用Django REST Framework构建生产级API
掌握序列化技术：实现复杂嵌套序列化、自定义字段与验证逻辑
设计认证体系：实现Token认证、JWT认证与OAuth2集成
构建权限系统：实现对象级权限、角色权限与自定义权限策略
实现高级功能：过滤、搜索、排序、分页、限流与版本控制
生成API文档：使用OpenAPI/Swagger自动生成交互式文档
掌握FastAPI：了解现代异步API框架的设计与使用

19.1 RESTful API设计理论

19.1.1 REST架构约束

REST（Representational State Transfer）定义了六项架构约束：

约束	描述	实践意义
客户端-服务器	关注点分离，UI与数据存储独立	前后端分离架构
无状态	每个请求包含全部必要信息	水平扩展能力
可缓存	响应必须明确标识是否可缓存	减少服务器负载
统一接口	资源标识、操作表述、自描述消息、HATEOAS	API一致性与可预测性
分层系统	客户端无法知道直接连接的是终端服务器	负载均衡、CDN
按需代码	服务器可扩展客户端功能（可选）	JavaScript小应用

19.1.2 RESTful URL设计

资源命名规范：

GET    /api/v1/posts          → 获取文章列表
POST   /api/v1/posts          → 创建文章
GET    /api/v1/posts/123      → 获取指定文章
PUT    /api/v1/posts/123      → 全量更新文章
PATCH  /api/v1/posts/123      → 部分更新文章
DELETE /api/v1/posts/123      → 删除文章

嵌套资源：

GET    /api/v1/posts/123/comments       → 获取文章评论
POST   /api/v1/posts/123/comments       → 创建评论
GET    /api/v1/posts/123/comments/456   → 获取指定评论

非CRUD操作：

POST   /api/v1/posts/123/publish        → 发布文章
POST   /api/v1/posts/123/like           → 点赞
POST   /api/v1/users/123/follow         → 关注用户

19.1.3 响应格式设计

json

{
    "code": 200,
    "message": "success",
    "data": {
        "id": 1,
        "title": "Django REST Framework入门",
        "content": "...",
        "author": {
            "id": 1,
            "username": "admin"
        },
        "tags": [
            {"id": 1, "name": "Python"},
            {"id": 2, "name": "DRF"}
        ],
        "created_at": "2024-01-15T10:30:00Z"
    }
}

分页响应：

json

{
    "code": 200,
    "message": "success",
    "data": {
        "items": [...],
        "pagination": {
            "page": 1,
            "per_page": 10,
            "total": 100,
            "pages": 10,
            "has_next": true,
            "has_prev": false
        }
    }
}

错误响应：

json

{
    "code": 422,
    "message": "Validation Error",
    "errors": [
        {
            "field": "title",
            "message": "标题长度不能少于5个字符"
        },
        {
            "field": "email",
            "message": "邮箱格式不正确"
        }
    ]
}

19.2 Django REST Framework

19.2.1 安装与配置

bash

pip install djangorestframework
pip install djangorestframework-simplejwt
pip install django-filter
pip install drf-yasg
pip install drf-spectacular

python

INSTALLED_APPS = [
    "rest_framework",
    "rest_framework_simplejwt",
    "django_filters",
    "drf_spectacular",
    "apps.api",
]

REST_FRAMEWORK = {
    "DEFAULT_AUTHENTICATION_CLASSES": [
        "rest_framework_simplejwt.authentication.JWTAuthentication",
        "rest_framework.authentication.SessionAuthentication",
    ],
    "DEFAULT_PERMISSION_CLASSES": [
        "rest_framework.permissions.IsAuthenticatedOrReadOnly",
    ],
    "DEFAULT_FILTER_BACKENDS": [
        "django_filters.rest_framework.DjangoFilterBackend",
        "rest_framework.filters.SearchFilter",
        "rest_framework.filters.OrderingFilter",
    ],
    "DEFAULT_PAGINATION_CLASS": "rest_framework.pagination.PageNumberPagination",
    "PAGE_SIZE": 20,
    "DEFAULT_SCHEMA_CLASS": "drf_spectacular.openapi.AutoSchema",
    "DEFAULT_RENDERER_CLASSES": [
        "rest_framework.renderers.JSONRenderer",
    ],
    "DEFAULT_PARSER_CLASSES": [
        "rest_framework.parsers.JSONParser",
    ],
    "EXCEPTION_HANDLER": "apps.api.exception_handlers.custom_exception_handler",
    "DEFAULT_THROTTLE_CLASSES": [
        "rest_framework.throttling.AnonRateThrottle",
        "rest_framework.throttling.UserRateThrottle",
    ],
    "DEFAULT_THROTTLE_RATES": {
        "anon": "100/hour",
        "user": "1000/hour",
    },
}

19.2.2 序列化器

基础序列化器：

python

from rest_framework import serializers
from .models import Post, Category, Tag, Comment


class TagSerializer(serializers.ModelSerializer):
    class Meta:
        model = Tag
        fields = ["id", "name", "slug"]


class CategorySerializer(serializers.ModelSerializer):
    post_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Category
        fields = ["id", "name", "slug", "description", "post_count"]


class CommentSerializer(serializers.ModelSerializer):
    author_name = serializers.CharField(source="author.username", read_only=True)
    replies = serializers.SerializerMethodField()

    class Meta:
        model = Comment
        fields = [
            "id", "content", "author_name", "parent",
            "is_approved", "created_at", "replies",
        ]
        read_only_fields = ["is_approved", "created_at"]

    def get_replies(self, obj):
        if obj.replies.exists():
            return CommentSerializer(
                obj.replies.filter(is_approved=True), many=True
            ).data
        return []


class PostListSerializer(serializers.ModelSerializer):
    author = serializers.CharField(source="author.username", read_only=True)
    category_name = serializers.CharField(source="category.name", read_only=True)
    tags = TagSerializer(many=True, read_only=True)
    comment_count = serializers.IntegerField(read_only=True)

    class Meta:
        model = Post
        fields = [
            "id", "title", "slug", "summary", "author",
            "category_name", "tags", "comment_count",
            "view_count", "published_at",
        ]


class PostDetailSerializer(serializers.ModelSerializer):
    author = serializers.CharField(source="author.username", read_only=True)
    category = CategorySerializer(read_only=True)
    category_id = serializers.PrimaryKeyRelatedField(
        queryset=Category.objects.all(), source="category", write_only=True,
        required=False,
    )
    tags = TagSerializer(many=True, read_only=True)
    tag_ids = serializers.PrimaryKeyRelatedField(
        queryset=Tag.objects.all(), source="tags", write_only=True,
        many=True, required=False,
    )
    comments = CommentSerializer(many=True, read_only=True)

    class Meta:
        model = Post
        fields = [
            "id", "title", "slug", "content", "summary",
            "status", "author", "category", "category_id",
            "tags", "tag_ids", "comments", "featured_image",
            "view_count", "is_featured", "published_at",
            "created_at", "updated_at",
        ]
        read_only_fields = ["view_count", "published_at", "created_at", "updated_at"]

    def validate_title(self, value):
        if len(value) < 5:
            raise serializers.ValidationError("标题长度不能少于5个字符")
        return value

    def validate(self, data):
        if data.get("status") == "PB" and not data.get("category"):
            raise serializers.ValidationError({
                "category_id": "发布文章必须选择分类"
            })
        return data

19.2.3 自定义字段与验证

python

class MarkdownField(serializers.Field):
    def to_representation(self, value):
        import markdown
        return markdown.markdown(value, extensions=["fenced_code", "toc"])

    def to_internal_value(self, data):
        if not isinstance(data, str):
            raise serializers.ValidationError("内容必须是字符串")
        return data


class PostSerializer(serializers.ModelSerializer):
    content_html = MarkdownField(source="content", read_only=True)

    class Meta:
        model = Post
        fields = ["id", "title", "content", "content_html"]


class BulkDeleteSerializer(serializers.Serializer):
    ids = serializers.ListField(
        child=serializers.IntegerField(),
        allow_empty=False,
        max_length=100,
    )

    def validate_ids(self, value):
        existing = set(Post.objects.filter(id__in=value).values_list("id", flat=True))
        missing = set(value) - existing
        if missing:
            raise serializers.ValidationError(f"以下ID不存在: {missing}")
        return value

19.2.4 ViewSet与路由

python

from rest_framework import viewsets, status, filters
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated, IsAuthenticatedOrReadOnly
from django_filters.rest_framework import DjangoFilterBackend
from .models import Post, Category, Tag, Comment
from .serializers import (
    PostListSerializer, PostDetailSerializer,
    CategorySerializer, TagSerializer, CommentSerializer,
)
from .permissions import IsAuthorOrReadOnly
from .filters import PostFilter


class PostViewSet(viewsets.ModelViewSet):
    queryset = Post.published.select_related(
        "author", "category"
    ).prefetch_related("tags", "comments").all()
    permission_classes = [IsAuthenticatedOrReadOnly, IsAuthorOrReadOnly]
    filter_backends = [DjangoFilterBackend, filters.SearchFilter, filters.OrderingFilter]
    filterset_class = PostFilter
    search_fields = ["title", "content"]
    ordering_fields = ["published_at", "view_count", "title"]
    ordering = ["-published_at"]
    lookup_field = "slug"

    def get_serializer_class(self):
        if self.action == "list":
            return PostListSerializer
        if self.action == "retrieve":
            return PostDetailSerializer
        return PostDetailSerializer

    def perform_create(self, serializer):
        serializer.save(author=self.request.user)

    @action(detail=True, methods=["post"], permission_classes=[IsAuthenticated])
    def like(self, request, slug=None):
        post = self.get_object()
        from .models import Like
        like, created = Like.objects.get_or_create(
            user=request.user, post=post
        )
        if not created:
            like.delete()
            return Response({"status": "unliked"})
        return Response({"status": "liked"}, status=status.HTTP_201_CREATED)

    @action(detail=True, methods=["post"], permission_classes=[IsAuthenticated])
    def publish(self, request, slug=None):
        post = self.get_object()
        if post.author != request.user and not request.user.is_staff:
            return Response(
                {"detail": "无权操作"}, status=status.HTTP_403_FORBIDDEN
            )
        post.status = Post.Status.PUBLISHED
        post.save()
        return Response({"status": "published"})

    @action(detail=False, methods=["post"], permission_classes=[IsAuthenticated])
    def bulk_delete(self, request):
        from .serializers import BulkDeleteSerializer
        serializer = BulkDeleteSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        deleted = Post.objects.filter(
            id__in=serializer.validated_data["ids"],
            author=request.user,
        ).delete()
        return Response({"deleted": deleted[0]}, status=status.HTTP_204_NO_CONTENT)


class CategoryViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = Category.objects.annotate(
        post_count=Count("posts")
    ).order_by("order", "name")
    serializer_class = CategorySerializer
    lookup_field = "slug"


class TagViewSet(viewsets.ReadOnlyModelViewSet):
    queryset = Tag.objects.all()
    serializer_class = TagSerializer
    lookup_field = "slug"


class CommentViewSet(viewsets.ModelViewSet):
    queryset = Comment.objects.select_related("author", "post").all()
    serializer_class = CommentSerializer
    permission_classes = [IsAuthenticatedOrReadOnly]

    def get_queryset(self):
        return super().get_queryset().filter(
            post__slug=self.kwargs["post_slug"],
            is_approved=True,
        )

    def perform_create(self, serializer):
        post = Post.objects.get(slug=self.kwargs["post_slug"])
        serializer.save(author=self.request.user, post=post)

19.2.5 路由配置

python

from rest_framework.routers import DefaultRouter
from rest_framework_nested import routers
from .views import PostViewSet, CategoryViewSet, TagViewSet, CommentViewSet

router = DefaultRouter()
router.register("posts", PostViewSet)
router.register("categories", CategoryViewSet)
router.register("tags", TagViewSet)

posts_router = routers.NestedDefaultRouter(router, "posts", lookup="post")
posts_router.register("comments", CommentViewSet, basename="post-comments")

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

19.3 认证系统

19.3.1 JWT认证

python

from rest_framework_simplejwt.views import TokenObtainPairView, TokenRefreshView
from rest_framework_simplejwt.serializers import TokenObtainPairSerializer


class CustomTokenObtainPairSerializer(TokenObtainPairSerializer):
    @classmethod
    def get_token(cls, user):
        token = super().get_token(user)
        token["username"] = user.username
        token["is_staff"] = user.is_staff
        return token

    def validate(self, attrs):
        data = super().validate(attrs)
        data["username"] = self.user.username
        data["email"] = self.user.email
        return data


class CustomTokenObtainPairView(TokenObtainPairView):
    serializer_class = CustomTokenObtainPairSerializer

JWT配置：

python

from datetime import timedelta

SIMPLE_JWT = {
    "ACCESS_TOKEN_LIFETIME": timedelta(minutes=30),
    "REFRESH_TOKEN_LIFETIME": timedelta(days=7),
    "ROTATE_REFRESH_TOKENS": True,
    "BLACKLIST_AFTER_ROTATION": True,
    "AUTH_HEADER_TYPES": ("Bearer",),
    "TOKEN_USER_CLASS": "apps.users.models.User",
}

URL配置：

python

from rest_framework_simplejwt.views import TokenVerifyView, TokenBlacklistView

urlpatterns = [
    path("auth/login/", CustomTokenObtainPairView.as_view(), name="token_obtain_pair"),
    path("auth/refresh/", TokenRefreshView.as_view(), name="token_refresh"),
    path("auth/verify/", TokenVerifyView.as_view(), name="token_verify"),
    path("auth/logout/", TokenBlacklistView.as_view(), name="token_blacklist"),
]

19.3.2 注册与密码重置

python

from rest_framework import serializers, status
from rest_framework.decorators import api_view, permission_classes
from rest_framework.permissions import AllowAny
from rest_framework.response import Response
from django.contrib.auth import get_user_model
from django.contrib.auth.tokens import default_token_generator
from django.utils.encoding import force_bytes
from django.utils.http import urlsafe_base64_encode, urlsafe_base64_decode

User = get_user_model()


class RegisterSerializer(serializers.ModelSerializer):
    password = serializers.CharField(write_only=True, min_length=8)
    confirm_password = serializers.CharField(write_only=True)

    class Meta:
        model = User
        fields = ["username", "email", "password", "confirm_password"]

    def validate(self, data):
        if data["password"] != data["confirm_password"]:
            raise serializers.ValidationError({"confirm_password": "密码不一致"})
        return data

    def create(self, validated_data):
        validated_data.pop("confirm_password")
        user = User.objects.create_user(**validated_data)
        return user


@api_view(["POST"])
@permission_classes([AllowAny])
def register_view(request):
    serializer = RegisterSerializer(data=request.data)
    serializer.is_valid(raise_exception=True)
    user = serializer.create(serializer.validated_data)
    return Response({
        "id": user.id,
        "username": user.username,
        "email": user.email,
    }, status=status.HTTP_201_CREATED)


@api_view(["POST"])
@permission_classes([AllowAny])
def password_reset_request(request):
    email = request.data.get("email")
    try:
        user = User.objects.get(email=email)
        uid = urlsafe_base64_encode(force_bytes(user.pk))
        token = default_token_generator.make_token(user)
        from .services import send_password_reset_email
        send_password_reset_email(user, uid, token)
        return Response({"message": "密码重置邮件已发送"})
    except User.DoesNotExist:
        return Response({"message": "密码重置邮件已发送"})


@api_view(["POST"])
@permission_classes([AllowAny])
def password_reset_confirm(request):
    uid = request.data.get("uid")
    token = request.data.get("token")
    new_password = request.data.get("new_password")

    try:
        user_id = urlsafe_base64_decode(uid).decode()
        user = User.objects.get(pk=user_id)
    except (TypeError, ValueError, User.DoesNotExist):
        return Response(
            {"error": "无效的重置链接"}, status=status.HTTP_400_BAD_REQUEST
        )

    if not default_token_generator.check_token(user, token):
        return Response(
            {"error": "重置链接已过期"}, status=status.HTTP_400_BAD_REQUEST
        )

    user.set_password(new_password)
    user.save()
    return Response({"message": "密码重置成功"})

19.4 权限系统

19.4.1 自定义权限

python

from rest_framework import permissions


class IsAuthorOrReadOnly(permissions.BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.method in permissions.SAFE_METHODS:
            return True
        return obj.author == 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):
    role_permissions = {
        "reader": ["list", "retrieve"],
        "author": ["list", "retrieve", "create", "update", "partial_update", "destroy"],
        "editor": ["list", "retrieve", "create", "update", "partial_update", "publish"],
        "admin": ["__all__"],
    }

    def has_permission(self, request, view):
        if not request.user.is_authenticated:
            return False

        action = view.action
        role = getattr(request.user, "role", "reader")
        allowed_actions = self.role_permissions.get(role, [])

        if "__all__" in allowed_actions:
            return True
        return action in allowed_actions


class IsOwnerOrAdmin(permissions.BasePermission):
    def has_object_permission(self, request, view, obj):
        if request.user.is_staff:
            return True
        owner_field = getattr(obj, "author", None) or getattr(obj, "user", None)
        return owner_field == request.user

19.5 过滤、搜索与排序

19.5.1 Django Filter

python

from django_filters import rest_framework as filters
from .models import Post


class PostFilter(filters.FilterSet):
    status = filters.ChoiceFilter(choices=Post.Status.choices)
    category = filters.CharFilter(field_name="category__slug")
    tag = filters.CharFilter(field_name="tags__slug")
    author = filters.NumberFilter(field_name="author__id")
    date_from = filters.DateTimeFilter(field_name="published_at", lookup_expr="gte")
    date_to = filters.DateTimeFilter(field_name="published_at", lookup_expr="lte")
    min_views = filters.NumberFilter(field_name="view_count", lookup_expr="gte")
    is_featured = filters.BooleanFilter()

    class Meta:
        model = Post
        fields = []

19.5.2 自定义分页

python

from rest_framework.pagination import PageNumberPagination, CursorPagination
from rest_framework.response import Response


class StandardPagination(PageNumberPagination):
    page_size = 20
    page_size_query_param = "per_page"
    max_page_size = 100

    def get_paginated_response(self, data):
        return Response({
            "items": data,
            "pagination": {
                "page": self.page.number,
                "per_page": self.get_page_size(self.request),
                "total": self.page.paginator.count,
                "pages": self.page.paginator.num_pages,
                "has_next": self.page.has_next(),
                "has_prev": self.page.has_previous(),
            },
        })


class CursorPostPagination(CursorPagination):
    ordering = "-published_at"
    page_size = 20
    cursor_query_param = "cursor"

19.6 异常处理

python

from rest_framework.views import exception_handler
from rest_framework.exceptions import APIException
from rest_framework import status


class BusinessError(APIException):
    status_code = status.HTTP_400_BAD_REQUEST
    default_detail = "业务处理失败"
    default_code = "business_error"


class ResourceLockedError(APIException):
    status_code = status.HTTP_423_LOCKED
    default_detail = "资源已被锁定"
    default_code = "resource_locked"


def custom_exception_handler(exc, context):
    response = exception_handler(exc, context)

    if response is not None:
        custom_data = {
            "code": response.status_code,
            "message": "请求处理失败",
            "errors": [],
        }

        if isinstance(response.data, dict):
            if "detail" in response.data:
                custom_data["message"] = str(response.data["detail"])
            else:
                for field, messages in response.data.items():
                    if isinstance(messages, list):
                        for msg in messages:
                            custom_data["errors"].append({
                                "field": field,
                                "message": str(msg),
                            })
                    else:
                        custom_data["errors"].append({
                            "field": field,
                            "message": str(messages),
                        })
        elif isinstance(response.data, list):
            custom_data["message"] = str(response.data[0])

        response.data = custom_data

    return response

19.7 API文档

19.7.1 drf-spectacular

python

from drf_spectacular.utils import extend_schema, extend_schema_view, OpenApiParameter, OpenApiExample
from drf_spectacular.types import OpenApiTypes


@extend_schema_view(
    list=extend_schema(
        summary="获取文章列表",
        description="返回已发布的文章列表，支持过滤、搜索和排序",
        parameters=[
            OpenApiParameter(
                name="category",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                description="按分类slug过滤",
            ),
            OpenApiParameter(
                name="search",
                type=OpenApiTypes.STR,
                location=OpenApiParameter.QUERY,
                description="搜索关键词",
            ),
        ],
        examples=[
            OpenApiExample(
                "成功响应示例",
                value={
                    "items": [{"id": 1, "title": "示例文章"}],
                    "pagination": {"page": 1, "total": 100},
                },
            ),
        ],
    ),
    create=extend_schema(
        summary="创建文章",
        description="创建一篇新文章，需要认证",
    ),
)
class PostViewSet(viewsets.ModelViewSet):
    ...

URL配置：

python

from drf_spectacular.views import SpectacularAPIView, SpectacularSwaggerView, SpectacularRedocView

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

19.8 FastAPI入门

19.8.1 FastAPI核心概念

FastAPI是现代高性能Python Web框架，基于Starlette和Pydantic构建：

bash

pip install fastapi uvicorn

python

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime

app = FastAPI(
    title="Blog API",
    version="1.0.0",
    description="基于FastAPI的博客API",
)


class PostCreate(BaseModel):
    title: str = Field(..., min_length=5, max_length=200)
    content: str = Field(..., min_length=10)
    category_id: Optional[int] = None
    tag_ids: list[int] = Field(default_factory=list)


class PostResponse(BaseModel):
    id: int
    title: str
    slug: str
    content: str
    author_name: str
    published_at: Optional[datetime]
    created_at: datetime

    model_config = {"from_attributes": True}


class PaginatedResponse(BaseModel):
    items: list[PostResponse]
    total: int
    page: int
    per_page: int
    pages: int


oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login")


async def get_current_user(token: str = Depends(oauth2_scheme)):
    from .auth import decode_token
    user = decode_token(token)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="无效的认证凭据",
        )
    return user


@app.get("/api/v1/posts", response_model=PaginatedResponse)
async def list_posts(
    page: int = 1,
    per_page: int = 20,
    category: Optional[str] = None,
    search: Optional[str] = None,
):
    from .services import get_posts
    return await get_posts(page, per_page, category, search)


@app.get("/api/v1/posts/{slug}", response_model=PostResponse)
async def get_post(slug: str):
    from .services import get_post_by_slug
    post = await get_post_by_slug(slug)
    if not post:
        raise HTTPException(status_code=404, detail="文章不存在")
    return post


@app.post("/api/v1/posts", response_model=PostResponse, status_code=201)
async def create_post(
    post_data: PostCreate,
    current_user=Depends(get_current_user),
):
    from .services import create_new_post
    return await create_new_post(post_data, current_user)


@app.put("/api/v1/posts/{slug}", response_model=PostResponse)
async def update_post(
    slug: str,
    post_data: PostCreate,
    current_user=Depends(get_current_user),
):
    from .services import update_existing_post
    post = await update_existing_post(slug, post_data, current_user)
    if not post:
        raise HTTPException(status_code=404, detail="文章不存在")
    return post


@app.delete("/api/v1/posts/{slug}", status_code=204)
async def delete_post(
    slug: str,
    current_user=Depends(get_current_user),
):
    from .services import delete_existing_post
    success = await delete_existing_post(slug, current_user)
    if not success:
        raise HTTPException(status_code=404, detail="文章不存在")

19.8.2 FastAPI依赖注入

python

from fastapi import Depends, Query
from typing import Annotated


class PaginationParams:
    def __init__(
        self,
        page: int = Query(1, ge=1, description="页码"),
        per_page: int = Query(20, ge=1, le=100, description="每页数量"),
    ):
        self.page = page
        self.per_page = per_page

    @property
    def offset(self):
        return (self.page - 1) * self.per_page


PaginationDep = Annotated[PaginationParams, Depends()]


@app.get("/api/v1/posts")
async def list_posts(pagination: PaginationDep):
    return {
        "page": pagination.page,
        "per_page": pagination.per_page,
        "offset": pagination.offset,
    }


class DatabaseSession:
    def __init__(self):
        self.session = None

    async def __aenter__(self):
        from .database import async_session
        self.session = async_session()
        return self.session

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self.session:
            await self.session.close()


async def get_db():
    async with DatabaseSession() as session:
        yield session


DbDep = Annotated[AsyncSession, Depends(get_db)]

19.9 API测试

19.9.1 DRF测试

python

from rest_framework.test import APITestCase
from django.contrib.auth import get_user_model
from django.urls import reverse
from ..models import Post, Category

User = get_user_model()


class PostAPITestCase(APITestCase):
    @classmethod
    def setUpTestData(cls):
        cls.user = User.objects.create_user(
            username="testuser", password="TestPass123!"
        )
        cls.category = Category.objects.create(name="Python", slug="python")
        cls.post = Post.objects.create(
            title="Test Post", slug="test-post",
            content="Test content", author=cls.user,
            category=cls.category, status=Post.Status.PUBLISHED,
        )

    def test_list_posts(self):
        url = reverse("api:post-list")
        response = self.client.get(url)
        self.assertEqual(response.status_code, 200)

    def test_retrieve_post(self):
        url = reverse("api:post-detail", kwargs={"slug": self.post.slug})
        response = self.client.get(url)
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data["title"], "Test Post")

    def test_create_post_unauthenticated(self):
        url = reverse("api:post-list")
        response = self.client.post(url, {
            "title": "New Post",
            "content": "New content",
        })
        self.assertEqual(response.status_code, 401)

    def test_create_post_authenticated(self):
        self.client.force_authenticate(user=self.user)
        url = reverse("api:post-list")
        response = self.client.post(url, {
            "title": "New Post",
            "content": "New content for the post",
            "category_id": self.category.id,
            "status": "PB",
        })
        self.assertEqual(response.status_code, 201)

    def test_update_post_by_author(self):
        self.client.force_authenticate(user=self.user)
        url = reverse("api:post-detail", kwargs={"slug": self.post.slug})
        response = self.client.patch(url, {"title": "Updated Title"})
        self.assertEqual(response.status_code, 200)
        self.assertEqual(response.data["title"], "Updated Title")

    def test_update_post_by_other_forbidden(self):
        other = User.objects.create_user(username="other", password="OtherPass123!")
        self.client.force_authenticate(user=other)
        url = reverse("api:post-detail", kwargs={"slug": self.post.slug})
        response = self.client.patch(url, {"title": "Hacked"})
        self.assertEqual(response.status_code, 403)

    def test_delete_post(self):
        self.client.force_authenticate(user=self.user)
        url = reverse("api:post-detail", kwargs={"slug": self.post.slug})
        response = self.client.delete(url)
        self.assertEqual(response.status_code, 204)

    def test_search_posts(self):
        url = reverse("api:post-list")
        response = self.client.get(url, {"search": "Test"})
        self.assertEqual(response.status_code, 200)

    def test_filter_by_category(self):
        url = reverse("api:post-list")
        response = self.client.get(url, {"category": "python"})
        self.assertEqual(response.status_code, 200)

19.9.2 FastAPI测试

python

import pytest
from httpx import AsyncClient
from app.main import app


@pytest.mark.asyncio
async def test_list_posts():
    async with AsyncClient(app=app, base_url="http://test") as client:
        response = await client.get("/api/v1/posts")
    assert response.status_code == 200


@pytest.mark.asyncio
async def test_create_post_authenticated():
    async with AsyncClient(app=app, base_url="http://test") as client:
        login_resp = await client.post("/auth/login", data={
            "username": "testuser",
            "password": "TestPass123!",
        })
        token = login_resp.json()["access_token"]

        response = await client.post(
            "/api/v1/posts",
            json={
                "title": "New Post",
                "content": "Content for the new post",
            },
            headers={"Authorization": f"Bearer {token}"},
        )
    assert response.status_code == 201

19.10 前沿技术动态

19.10.1 API设计趋势

GraphQL：灵活的数据查询语言，客户端按需获取字段
gRPC：基于Protocol Buffers的高性能RPC框架
Server-Sent Events（SSE）：服务器向客户端单向推送
WebSocket：全双工实时通信
API Gateway：微服务架构中的统一入口

19.10.2 OpenAPI 3.1与JSON Schema

OpenAPI 3.1完全兼容JSON Schema Draft 2020-12，支持更精确的类型定义：

python

from drf_spectacular.utils import extend_schema, inline_serializer

@extend_schema(
    responses={
        200: inline_serializer(
            name="PostStatsResponse",
            fields={
                "total_posts": serializers.IntegerField(),
                "avg_views": serializers.FloatField(),
                "top_categories": CategorySerializer(many=True),
            },
        )
    }
)
@api_view(["GET"])
def post_stats(request):
    ...

19.11 本章小结

本章系统阐述了Python API开发的核心知识体系：

REST理论：六项架构约束、URL设计规范、响应格式标准
DRF框架：序列化器、ViewSet、路由与配置体系
高级序列化：嵌套序列化、自定义字段、批量操作
认证系统：JWT令牌认证、注册与密码重置
权限系统：对象级权限、角色权限与自定义策略
查询优化：Django Filter过滤、搜索、排序与自定义分页
异常处理：统一错误响应格式与自定义异常
API文档：drf-spectacular自动生成OpenAPI文档
FastAPI：现代异步框架、Pydantic验证与依赖注入
API测试：DRF测试与FastAPI异步测试

19.12 习题与项目练习

基础题

使用DRF创建一个图书管理API，支持图书的CRUD操作，包含作者和分类的嵌套序列化。
实现JWT认证系统，包含用户注册、登录、令牌刷新和退出功能。
为API添加过滤功能，支持按分类、标签、日期范围和关键词搜索。

进阶题

实现一个完整的评论API，支持嵌套回复、评论审核和评论通知，使用嵌套路由设计。
设计并实现API限流系统，要求对不同用户角色（匿名、普通用户、VIP）设置不同的请求频率限制。
使用FastAPI实现一个文件上传API，支持图片和文档上传，包含文件类型验证、大小限制和缩略图生成。

综合项目

社交平台API：构建一个完整的社交平台API，包含：
- 用户注册/登录（JWT + OAuth2）
- 动态发布/评论/点赞
- 用户关注系统
- 实时消息通知（WebSocket）
- 全文搜索
- API文档与版本控制
- 限流与缓存
电商API：构建一个电商API，包含：
- 商品管理（SPU/SKU）
- 购物车与订单
- 支付集成
- 库存管理
- 数据统计API
- 管理后台API

思考题

RESTful API与GraphQL在数据获取效率、开发体验和性能方面各有何优劣？在什么场景下应选择哪种方案？请从过度获取（Over-fetching）、不足获取（Under-fetching）和N+1问题三个维度分析。
在微服务架构中，API Gateway如何实现请求路由、认证转发、限流熔断和协议转换？请设计一个基于Python的API Gateway方案。

19.13 延伸阅读

19.13.1 API设计规范

RESTful Web Services (Leonard Richardson) — REST设计原则
《Web API Design》 (Apigee) — API设计最佳实践
OpenAPI Specification (https://spec.openapis.org/oas/latest.html) — API规范标准

19.13.2 Django REST Framework

DRF官方文档 (https://www.django-rest-framework.org/) — DRF权威指南
drf-spectacular (https://drf-spectacular.readthedocs.io/) — OpenAPI文档生成
django-rest-framework-simplejwt (https://django-rest-framework-simplejwt.readthedocs.io/) — JWT认证

19.13.3 FastAPI生态

FastAPI官方文档 (https://fastapi.tiangolo.com/) — FastAPI教程
Pydantic (https://docs.pydantic.dev/) — 数据验证
Uvicorn (https://www.uvicorn.org/) — ASGI服务器

19.13.4 API安全与性能

OWASP API Security Top 10 (https://owasp.org/www-project-api-security/) — API安全指南
GraphQL (https://graphql.org/learn/) — GraphQL教程
gRPC (https://grpc.io/docs/) — 高性能RPC框架

下一章：第20章 Tkinter GUI开发

第19章 API开发 ​

学习目标 ​

19.1 RESTful API设计理论 ​

19.1.1 REST架构约束 ​

19.1.2 RESTful URL设计 ​

19.1.3 响应格式设计 ​

19.2 Django REST Framework ​

19.2.1 安装与配置 ​

19.2.2 序列化器 ​

19.2.3 自定义字段与验证 ​

19.2.4 ViewSet与路由 ​

19.2.5 路由配置 ​

19.3 认证系统 ​

19.3.1 JWT认证 ​

19.3.2 注册与密码重置 ​

19.4 权限系统 ​

19.4.1 自定义权限 ​

19.5 过滤、搜索与排序 ​

19.5.1 Django Filter ​

19.5.2 自定义分页 ​

19.6 异常处理 ​

19.7 API文档 ​

19.7.1 drf-spectacular ​

19.8 FastAPI入门 ​

19.8.1 FastAPI核心概念 ​

19.8.2 FastAPI依赖注入 ​

19.9 API测试 ​

19.9.1 DRF测试 ​

19.9.2 FastAPI测试 ​

19.10 前沿技术动态 ​

19.10.1 API设计趋势 ​

19.10.2 OpenAPI 3.1与JSON Schema ​

19.11 本章小结 ​

19.12 习题与项目练习 ​

基础题 ​

进阶题 ​

综合项目 ​

思考题 ​

19.13 延伸阅读 ​

19.13.1 API设计规范 ​

19.13.2 Django REST Framework ​

19.13.3 FastAPI生态 ​

19.13.4 API安全与性能 ​

第19章 API开发

学习目标

19.1 RESTful API设计理论

19.1.1 REST架构约束

19.1.2 RESTful URL设计

19.1.3 响应格式设计

19.2 Django REST Framework

19.2.1 安装与配置

19.2.2 序列化器

19.2.3 自定义字段与验证

19.2.4 ViewSet与路由

19.2.5 路由配置

19.3 认证系统

19.3.1 JWT认证

19.3.2 注册与密码重置

19.4 权限系统

19.4.1 自定义权限

19.5 过滤、搜索与排序

19.5.1 Django Filter

19.5.2 自定义分页

19.6 异常处理

19.7 API文档

19.7.1 drf-spectacular

19.8 FastAPI入门

19.8.1 FastAPI核心概念

19.8.2 FastAPI依赖注入

19.9 API测试

19.9.1 DRF测试

19.9.2 FastAPI测试

19.10 前沿技术动态

19.10.1 API设计趋势

19.10.2 OpenAPI 3.1与JSON Schema

19.11 本章小结

19.12 习题与项目练习

基础题

进阶题

综合项目

思考题

19.13 延伸阅读

19.13.1 API设计规范

19.13.2 Django REST Framework

19.13.3 FastAPI生态

19.13.4 API安全与性能