引言
官网地址：https://fastapi.tiangolo.com/
Github地址：https://github.com/fastapi/fastapi
资料地址：https://mp.weixin.qq.com/s/ECNGeoDXAcDTXZ05lUEh3g
FastAPI是一个基于Python的现代Web框架，专为构建高性能API而设计。
FastAPI官方文档总结了它的核心能力：
- 高性能：基于Starlette（异步Web框架）和Pydantic（高性能数据校验），在Python Web框架中属于性能天花板。
- 自动生成API文档：无需编写文档，自动生成Swagger UI（/docs）和ReDoc（/redoc）。
- 使用Python类型提示自动校验参数：自动请求参数校验、自动响应模型校验、IDE自动补全体验极佳。
- 原生异步支持：完全支持async/await，适合高并发I/O场景。
- 依赖注入系统：支持权限校验、Token校验、DB会话管理、统一行为注入。
- 生产级特性：内置CORS、GZip、HTTPS重定向等中间件，支持WebSocket实时通信。
- 易维护：类型提示 + 自动补全，适合微服务架构。
传统的Python Web框架（如Flask、Django）基于WSGI（Web Server Gateway Interface）规范。WSGI是同步的——每个请求独占一个线程，直到处理完才能释放。这就好比一个餐厅里，每个服务员一次只能服务一桌客人，其他客人只能干等着。而FastAPI基于ASGI（Asynchronous Server Gateway Interface）规范。ASGI是异步非阻塞的——每个请求在事件循环中被调度，而不是独占线程资源。这就是为什么FastAPI能够轻松处理数千个并发连接。
三引擎驱动架构：
- ① 路由系统：基于路径操作装饰器（
@app.get、@app.post）实现RESTful路由，支持路径参数和查询参数的自动解析。其路径匹配算法采用正则表达式优化，在路径参数较多时比Flask的Werkzeug路由性能提升达40%。 - ② 依赖注入系统：通过
Depends关键字实现服务依赖的自动解析，特别适合数据库连接等资源的统一管理。其底层实现采用函数装饰器模式，通过__wrapped__属性保留原始函数，在运行时动态注入依赖项。 - ③ 数据验证引擎：FastAPI的数据验证基于Pydantic的
BaseModel，验证流程包含字段类型检查、约束条件验证、嵌套模型验证和额外属性检查。
正文
环境搭建
搭建环境（使用虚拟环境管理版本）：
# 创建项目目录
mkdir fastapi-demo
cd fastapi-demo
# 创建虚拟环境（推荐）
python3 -m venv venv
source venv/bin/activate # Windows: venv\Scripts\activate
# 安装 FastAPI 和 Uvicorn （Uvicorn是一个ASGI服务器）
pip install fastapi uvicorn[standard]
启动服务
启动服务：
uvicorn main:app --reload --host 0.0.0.0 --port 8000
参数说明：
main:app —— main.py文件中的app实例
--reload —— 开发模式下自动重启（生产环境不要用）
--host —— 监听地址
--port —— 端口号
启动后，访问以下地址：
API服务：http://localhost:8000
Swagger文档：http://localhost:8000/docs
ReDoc文档：http://localhost:8000/redoc
路径参数与查询参数
from fastapi import FastAPI
app = FastAPI()
# 路径参数：从URL路径中提取
@app.get("/users/{user_id}")
asyncdef get_user(user_id: int):# 自动类型转换 + 校验
return {"user_id": user_id, "name": f"User_{user_id}"}
# 查询参数：从URL问号后面提取
@app.get("/items")
asyncdef list_items(
skip: int = 0, # 默认值
limit: int = 10, # 默认值
category: str | None = None # 可选参数
):
return {"skip": skip, "limit": limit, "category": category}
代码解读：
- 路径参数{user_id}会从URL中提取，user_id: int会自动进行类型转换和校验——传入非数字会返回422错误
- 查询参数从?skip=0&limit=10中提取，有默认值的参数是可选的
- 类型注解让IDE能提供自动补全，也让框架能自动校验
访问示例：
GET /users/123 → {"user_id": 123, "name": "User_123"}
GET /items?skip=5&limit=20&category=books
请求体与Pydantic模型
Pydantic验证流程，当一个请求到达FastAPI时，数据验证的流程是这样的：
用Pydantic模型定义请求和响应的数据结构。
from fastapi import FastAPI
from pydantic import BaseModel, Field, EmailStr
from typing import Optional
from datetime import datetime
app = FastAPI()
# 定义请求体模型
class UserCreate(BaseModel):
username: str = Field(..., min_length=3, max_length=20, description="用户名")
email: EmailStr = Field(..., description="邮箱地址")
password: str = Field(..., min_length=8, description="密码")
age: Optional[int] = Field(None, ge=0, le=150, description="年龄")
tags: list[str] = []
# 定义响应体模型
class UserResponse(BaseModel):
id: int
username: str
email: str
age: Optional[int]
created_at: datetime
@app.post("/users", response_model=UserResponse)
asyncdef create_user(user: UserCreate):
# 业务逻辑：创建用户
return UserResponse(
id=1,
username=user.username,
email=user.email,
age=user.age,
created_at=datetime.now()
)
代码解读：
- UserCreate定义了请求体的结构，Field提供了额外的校验规则（最小长度、最大长度、取值范围等）
- EmailStr会自动校验邮箱格式
- response_model=UserResponse指定了响应的数据结构，框架会自动过滤掉不在模型中的字段
- 如果请求缺少必填字段或字段类型不对，FastAPI会自动返回422错误并说明原因
这就是“声明式编程”的魅力——你只需要声明“我要什么”，框架帮你处理“怎么校验”。
依赖注入
FastAPI的依赖注入系统非常灵活，适合处理权限校验、数据库会话管理等横切关注点。
from fastapi import FastAPI, Depends, Header, HTTPException
app = FastAPI()
# 定义一个依赖：验证Token
asyncdef verify_token(authorization: str = Header(...)):
"""从请求头中提取并验证Token"""
ifnot authorization.startswith("Bearer "):
raise HTTPException(status_code=401, detail="无效的认证格式")
token = authorization.replace("Bearer ", "")
if token != "valid-token":
raise HTTPException(status_code=401, detail="无效的Token")
return {"user_id": 1, "username": "admin"}
# 使用依赖
@app.get("/protected")
asyncdef protected_route(user: dict = Depends(verify_token)):
return {"message": f"欢迎, {user['username']}!", "user": user}
代码解读：
verify_token是一个依赖函数，从请求头中提取Authorization并验证
Depends(verify_token)将依赖注入到路由函数中
如果验证失败，自动返回401错误
验证通过后，返回的用户信息会作为user参数传入路由函数
这种设计让认证逻辑和业务逻辑完全分离，代码更清晰、更可测试。
异步支持
FastAPI原生支持async/await，这是它高性能的关键。
import asyncio
from fastapi import FastAPI
app = FastAPI()
# 同步函数（适用于CPU密集型操作）
@app.get("/sync")
def sync_endpoint():
# 同步操作，会阻塞线程
return {"result": "done"}
# 异步函数（适用于I/O密集型操作）
@app.get("/async")
asyncdef async_endpoint():
# 模拟I/O操作：数据库查询、外部API调用等
await asyncio.sleep(1)
# 在等待期间，事件循环可以处理其他请求
return {"result": "done after 1 second"}
# 混合使用：在异步函数中调用同步代码
@app.get("/mixed")
asyncdef mixed_endpoint():
# 使用 run_in_executor 将同步代码放到线程池执行
result = await asyncio.to_thread(sync_heavy_work)
return {"result": result}
def sync_heavy_work():
# CPU密集型操作
return sum(range(1000000))
代码解读：
异步函数用async def定义，在等待I/O时释放线程资源
await asyncio.sleep(1)模拟I/O等待，期间事件循环可以处理其他请求
对于CPU密集型操作，使用asyncio.to_thread放到线程池执行，避免阻塞事件循环
数据库集成
异步SQLAlchemy集成
from fastapi import FastAPI, Depends
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession, async_sessionmaker
from sqlalchemy.orm import declarative_base, Mapped, mapped_column
from sqlalchemy import select
# 数据库配置
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/db"
engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
Base = declarative_base()
# 定义模型
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
username: Mapped[str] = mapped_column(unique=True)
email: Mapped[str]
# 依赖：获取数据库会话
asyncdef get_db():
asyncwith AsyncSessionLocal() as session:
yield session
app = FastAPI()
@app.get("/users")
asyncdef get_users(db: AsyncSession = Depends(get_db)):
result = await db.execute(select(User))
users = result.scalars().all()
return [{"id": u.id, "username": u.username, "email": u.email} for u in users]
代码解读：
create_async_engine创建异步数据库引擎
AsyncSessionLocal是异步会话工厂
get_db是一个依赖，每个请求创建一个数据库会话，请求结束后自动关闭
所有数据库操作都是异步的，不会阻塞事件循环
统一响应与异常处理
在生产环境中，统一的响应格式和异常处理是必不可少的。
统一响应模型
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Generic, TypeVar, Optional
T = TypeVar("T")
class ApiResponse(BaseModel, Generic[T]):
"""统一API响应格式"""
code: int = 200
msg: str = "success"
data: Optional[T] = None
app = FastAPI()
@app.get("/users/{user_id}", response_model=ApiResponse)
asyncdef get_user(user_id: int):
if user_id <= 0:
return ApiResponse(code=400, msg="用户ID必须大于0")
# 模拟查询
user = {"id": user_id, "name": f"User_{user_id}"}
return ApiResponse(data=user)
全局异常处理
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
app = FastAPI()
@app.exception_handler(HTTPException)
asyncdef http_exception_handler(request: Request, exc: HTTPException):
return JSONResponse(
status_code=exc.status_code,
content={"code": exc.status_code, "msg": exc.detail, "data": None}
)
@app.exception_handler(Exception)
asyncdef general_exception_handler(request: Request, exc: Exception):
return JSONResponse(
status_code=500,
content={"code": 500, "msg": "服务器内部错误", "data": None}
)
有了全局异常处理，任何未捕获的异常都会被统一格式化为规范的响应结构，前端对接时再也不用猜“这个接口返回的格式是什么”了。
中间件
中间件可以在请求进入路由之前或响应返回之前进行统一处理。
from fastapi import FastAPI, Request
import time
app = FastAPI()
# 日志中间件：记录每个请求的耗时
@app.middleware("http")
asyncdef log_requests(request: Request, call_next):
start_time = time.time()
# 记录请求信息
print(f"收到请求: {request.method} {request.url.path}")
# 继续处理请求
response = await call_next(request)
# 记录响应信息
process_time = time.time() - start_time
print(f"请求完成: {process_time:.4f}秒")
# 在响应头中添加处理时间
response.headers["X-Process-Time"] = str(process_time)
return response
自动文档
FastAPI会根据路由定义、Pydantic模型和类型注解，自动生成两份文档：
- Swagger UI（/docs）：交互式文档，可以在线调试API
- ReDoc（/redoc）：更美观的静态文档
启动服务后访问/docs，你会看到一份完整的、可交互的API文档——参数说明、请求示例、响应示例全部自动生成。