当前位置：首页 > news >正文

【框架篇二】FastAPI路由与请求处理

news 2025/8/22 16:17:52

一、FastAPI架构与工作原理

1.1 核心组件

1.2 流程图

1.3 路由定义

二、参数处理的完整体系

2.1 参数类型图

2.2 路径参数用法

2.3 查询参数用法

2.4 请求体处理实例

三、响应处理与数据序列化

3.1 流程图

3.2 设计原则

3.3 HTTP状态码

3.3.1 成功 (2xx)

3.3.2 重定向 (3xx)

3.3.3 客户端错误 (4xx)

3.3.4 服务器错误 (5xx)

四、文件处理

4.1 流程图

4.2 文件上传实现

总结

路由系统就像是Web应用的"神经网络"，决定了当用户发送不同的HTTP请求时，应用程序如何响应和处理。FastAPI的路由系统不仅功能强大，还具有出色的性能和开发体验。让我们一起深入了解这个现代Python Web开发的核心组件。

一、FastAPI架构与工作原理

FastAPI的路由系统建立在Starlette框架之上，采用了基于 装饰器 的声明式路由定义方式。这种设计让代码更加直观和易于维护，同时保持了高性能的特性。

1.1 核心组件

FastAPI的路由系统由几个核心组件构成：

路由装饰器：如@app.get()、@app.post()等。这些装饰器不仅定义了HTTP方法，还可以指定路径、响应模型、状态码等多种参数。

路径操作函数：被装饰器修饰的Python函数，负责处理具体的业务逻辑。FastAPI会自动解析函数签名，提取参数信息，并进行类型验证。

路由表：FastAPI内部维护的路由映射表，记录了URL模式与处理函数的对应关系。当请求到来时，FastAPI会遍历这个表找到匹配的路由。

参数解析器：负责从HTTP请求中提取各种参数，包括路径参数、查询参数、请求头、请求体等，并根据类型注解进行自动转换和验证。

1.2 流程图

1.3 路由定义

from fastapi import FastAPI
from typing import Optionalapp = FastAPI(title="现代Web API", version="1.0.0")# 最简单的GET路由
@app.get("/")
def read_root():"""根路径，返回欢迎信息"""return {"message": "欢迎使用FastAPI", "version": "1.0.0"}# 带路径参数的路由
@app.get("/users/{user_id}")
def get_user(user_id: int):"""根据用户ID获取用户信息"""return {"user_id": user_id,"name": f"用户{user_id}","status": "active"}# 支持多种HTTP方法
@app.post("/users/")
def create_user():return {"message": "用户创建成功"}@app.put("/users/{user_id}")
def update_user(user_id: int):return {"message": f"用户{user_id}更新成功"}@app.delete("/users/{user_id}")
def delete_user(user_id: int):return {"message": f"用户{user_id}删除成功"}

FastAPI的路由系统具有以下显著优势：

自动类型转换：当你在函数参数中指定类型注解时，FastAPI会自动进行类型转换。如果转换失败，会返回详细的错误信息。

路径优先级：FastAPI按照路由定义的顺序进行匹配，更具体的路径应该定义在更通用的路径之前。

自动文档生成：每个路由都会自动出现在Swagger UI文档中，包括参数说明、响应格式等信息。

二、参数处理的完整体系

FastAPI提供了一套完整的参数处理体系，能够优雅地处理Web开发中遇到的各种参数类型。这个体系不仅功能强大，还具有出色的开发体验和类型安全性。

2.1 参数类型图

2.2 路径参数用法

路径参数是URL路径的一部分，它们为API提供了动态性和灵活性。FastAPI的路径参数支持多种数据类型和验证规则。

from fastapi import Path
from enum import Enum# 枚举类型的路径参数
class ItemType(str, Enum):electronics = "electronics"clothing = "clothing"books = "books"@app.get("/items/{item_type}/{item_id}")
def get_item(item_type: ItemType,item_id: int = Path(..., gt=0, description="商品ID，必须大于0")
):"""获取特定类型的商品信息- item_type: 商品类型，只能是 electronics, clothing, books 之一- item_id: 商品ID，必须是正整数"""return {"item_type": item_type,"item_id": item_id,"message": f"获取{item_type.value}类型的第{item_id}号商品"}# 复杂路径参数验证
@app.get("/users/{user_id}/posts/{post_id}")
def get_user_post(user_id: int = Path(..., ge=1, le=999999, description="用户ID"),post_id: int = Path(..., ge=1, description="文章ID")
):"""获取用户的特定文章"""return {"user_id": user_id,"post_id": post_id,"title": f"用户{user_id}的第{post_id}篇文章","content": "这里是文章内容..."}

2.3 查询参数用法

查询参数提供了丰富的过滤和配置选项，是构建灵活API的重要工具。

from fastapi import Query
from typing import List, Optional
from datetime import datetime@app.get("/search/")
def search_items(q: Optional[str] = Query(None, min_length=1, max_length=50, description="搜索关键词"),category: Optional[str] = Query(None, regex="^[a-zA-Z]+$", description="商品类别"),tags: List[str] = Query([], description="标签列表"),min_price: Optional[float] = Query(None, ge=0, description="最低价格"),max_price: Optional[float] = Query(None, ge=0, description="最高价格"),limit: int = Query(10, ge=1, le=100, description="返回结果数量"),offset: int = Query(0, ge=0, description="跳过的结果数量"),sort_by: Optional[str] = Query("created_at", regex="^(name|price|created_at)$"),order: Optional[str] = Query("desc", regex="^(asc|desc)$")
):"""高级商品搜索API支持多种过滤条件：- 关键词搜索- 类别筛选- 标签过滤- 价格区间- 分页和排序"""# 构建搜索条件filters = {}if q:filters["keyword"] = qif category:filters["category"] = categoryif tags:filters["tags"] = tagsif min_price is not None:filters["min_price"] = min_priceif max_price is not None:filters["max_price"] = max_pricereturn {"filters": filters,"pagination": {"limit": limit, "offset": offset},"sorting": {"sort_by": sort_by, "order": order},"results": f"模拟搜索结果，共找到 {limit} 条记录"}

2.4 请求体处理实例

请求体是POST、PUT等请求传递复杂数据的主要方式。FastAPI结合Pydantic提供了强大的请求体处理能力。

from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional, List
from datetime import datetimeclass Address(BaseModel):"""地址信息模型"""street: str = Field(..., min_length=1, max_length=100)city: str = Field(..., min_length=1, max_length=50)country: str = Field(..., min_length=2, max_length=50)postal_code: str = Field(..., regex=r"^\d{6}$")class UserCreate(BaseModel):"""用户创建模型"""username: str = Field(..., min_length=3, max_length=20, regex="^[a-zA-Z0-9_]+$")email: EmailStrpassword: str = Field(..., min_length=8, max_length=100)full_name: Optional[str] = Field(None, max_length=50)age: Optional[int] = Field(None, ge=13, le=120)interests: List[str] = Field(default_factory=list, max_items=10)address: Optional[Address] = Noneis_active: bool = True@validator('password')def validate_password(cls, v):"""密码强度验证"""if not any(c.isupper() for c in v):raise ValueError('密码必须包含至少一个大写字母')if not any(c.islower() for c in v):raise ValueError('密码必须包含至少一个小写字母')if not any(c.isdigit() for c in v):raise ValueError('密码必须包含至少一个数字')return v@validator('interests')def validate_interests(cls, v):"""兴趣爱好验证"""if len(v) != len(set(v)):raise ValueError('兴趣爱好不能重复')return v@app.post("/users/")
def create_user(user: UserCreate):"""创建新用户支持完整的用户信息，包括：- 基本信息验证- 密码强度检查- 地址信息- 兴趣爱好管理"""return {"message": "用户创建成功","user_id": 12345,"username": user.username,"created_at": datetime.now().isoformat()}

三、响应处理与数据序列化

响应处理是API开发中的关键环节，决定了客户端接收到的数据格式和结构。FastAPI提供了灵活而强大的响应处理机制，能够满足各种复杂的业务需求。