FastAPI 基础知识-尧图网站建设

📅 发布时间：2026/7/26 22:07:28

1. FastAPI 是什么

FastAPI 是一个现代、高性能的 Python Web 框架，专为构建 API 而生。它在 Python 3.6+ 基础上利用类型注解（type hints），自动完成数据校验、序列化、文档生成。

核心特性

极高性能

基于 Starlette + Pydantic，性能与 Node.js / Go 相当。异步原生，天然支持高并发

Pydantic 深度集成

用 Python 类型注解定义数据模型 → 自动校验、转换、序列化。零额外代码

自动生成文档

开箱即用 Swagger UI (/docs) + ReDoc (/redoc)，交互式测试你的 API

依赖注入

内置优雅的依赖注入系统，DB session、认证、权限可复用、可单元测试

与其他框架对比

特性	FastAPI	Flask	Django REST
性能	⭐ 极快（异步）	一般（同步）	中等（同步）
类型校验	✅ 内置(Pydantic)	❌ 需手动/Marshmallow	⚠️ Serializer
自动文档	✅ Swagger+ReDoc	❌ Flask-RESTx 等	✅ DRF-YASG
异步	✅ 原生异步	⚠️ 额外插件	⚠️ 3.1+ 异步
依赖注入	✅ 内置	❌ 无	❌ 无
学习曲线	低（Pythonic）	低	中高

2. 环境搭建与第一个 API

安装

bash
# 推荐用 uv 或 pip
$ pip install fastapi uvicorn# 安装完整生态
$ pip install fastapi[all]
# 等价于: fastapi + uvicorn + pydantic + python-multipart + ...# 验证安装
$ python -c "import fastapi; print(fastapi.__version__)"

第一个 API：Hello World

Python
# main.py — 最小 FastAPI 应用
from fastapi import FastAPI# 创建应用实例
app = FastAPI(title="我的第一个 API",description="FastAPI 入门示例",version="0.1.0",
)@app.get("/")
async def root():return {"message": "Hello, FastAPI!"}@app.get("/ping")
def ping():return {"status": "ok", "timestamp": "2026-07-01T00:00:00Z"}

bash
# 启动服务器
$ uvicorn main:app --reload --host 0.0.0.0 --port 8000# 浏览器打开: http://127.0.0.1:8000
# 自动文档:     http://127.0.0.1:8000/docs
# ReDoc:        http://127.0.0.1:8000/redoc
# OpenAPI JSON: http://127.0.0.1:8000/openapi.json

 关键概念： 
• async def 定义异步路由（非阻塞 I/O 场景首选） 
• def 定义同步路由（CPU 密集任务用线程池运行） 
• 路径操作装饰器：@app.get / @app.post / @app.put / @app.delete / @app.patch 
• --reload 开发模式，修改代码后自动重启

3. Pydantic 模型详解

Pydantic 是 FastAPI 的数据层灵魂——请求体校验、响应序列化、配置管理全都基于它。FastAPI 内部集成了 Pydantic v2（Rust 核心，性能比 v1 快数倍）。

3.1 基础模型

Python
from pydantic import BaseModel
from datetime import datetime
from typing import Optionalclass User(BaseModel):id: intname: stremail: strage: int = 18              # 默认值is_active: bool = Truecreated_at: Optional[datetime] = None  # 可选字段tags: list[str] = []         # 列表类型# 使用示例
user = User(id=1,name="张三",email="zhang@example.com",age=25,tags=["admin", "developer"],
)
print(user.model_dump())       # → dict，v2 方法
print(user.model_dump_json())  # → JSON 字符串
print(user.name)              # → 直接访问属性

3.2 字段校验与约束

Python
from pydantic import BaseModel, Field, EmailStr, field_validator
from typing import Optionalclass CreateUserRequest(BaseModel):# Field 约束name: str = Field(..., min_length=2, max_length=50, description="用户名")email: EmailStr                   # pip install pydantic[email]age: int = Field(..., ge=0, le=150, description="年龄 0-150")password: str = Field(..., min_length=8)phone: Optional[str] = Field(None, pattern=r"^1[3-9]\d{9}$")# 自定义校验器（v2 语法）@field_validator("name")@classmethoddef validate_name(cls, v):if "admin" in v.lower():raise ValueError("用户名不能包含 admin")return v.strip()@field_validator("password")@classmethoddef validate_password(cls, v):if not any(c.isupper() for c in v):raise ValueError("密码必须包含大写字母")if not any(c.isdigit() for c in v):raise ValueError("密码必须包含数字")return v# 测试校验
try:u = CreateUserRequest(name=" 李四 ",email="lisi@test.com",age=25,password="Abc12345",phone="13800138000",)print("✅ 校验通过:", u.name)  # 自动 strip → "李四"
except Exception as e:print("❌", e)

3.3 嵌套模型与配置

Python
from pydantic import BaseModel, ConfigDict
from datetime import datetime
from typing import List, Optionalclass Address(BaseModel):city: strstreet: strzip_code: str = Field("", pattern=r"^\d{6}$")class Role(BaseModel):name: strpermissions: List[str] = []class Employee(BaseModel):# 配置：允许从 ORM 对象创建（后面 SQLAlchemy 用）model_config = ConfigDict(from_attributes=True)id: intname: straddress: Address              # 嵌套模型roles: List[Role] = []        # 模型列表created_at: datetime# 构造嵌套数据
emp = Employee(id=1,name="王五",address={                    # 自动从 dict 转换"city": "北京","street": "中关村大街","zip_code": "100080",},roles=[{"name": "管理员", "permissions": ["read", "write"]},],created_at="2026-07-01T10:00:00",  # 字符串自动转 datetime
)
print(emp.model_dump_json(indent=2))

3.4 Pydantic 在 FastAPI 中的位置

数据流HTTP 请求 (JSON)↓FastAPI 路由   ← 自动校验请求数据↓Pydantic 模型  ← 类型转换、校验、字段约束↓你的业务逻辑    ← 操作模型对象↓Pydantic 模型  ← model_dump() 序列化↓JSON 响应 (自动)↓HTTP 响应

4. 请求参数处理

FastAPI 支持5 种参数来源，全通过类型注解自动解析。

4.1 路径参数

Python
@app.get("/users/{user_id}")
async def get_user(user_id: int):    # 自动转换 int + 校验return {"user_id": user_id, "message": "用户信息"}@app.get("/items/{item_id}/versions/{version}")
async def get_item_version(item_id: int,version: int = Path(..., ge=1),  # 额外校验
):return {"item_id": item_id, "version": version}

4.2 查询参数

Python
from fastapi import Query@app.get("/users")
async def list_users(page: int = Query(1, ge=1, description="页码"),size: int = Query(10, ge=1, le=100, description="每页条数"),keyword: str | None = Query(None, min_length=2, description="搜索关键字"),sort_by: str = Query("id", pattern=r"^(id|name|age|created_at)$"),order: str = Query("asc", pattern=r"^(asc|desc)$"),
):return {"page": page,"size": size,"keyword": keyword,"sort_by": sort_by,"order": order,"items": [],  # 省略查询}# 请求示例:
# GET /users?page=1&size=20&keyword=张三&sort_by=age&order=desc

4.3 请求体（Body）—— 核心玩法

Python
from fastapi import Bodyclass CreateUserReq(BaseModel):name: stremail: strage: int = Field(18, ge=0)@app.post("/users", status_code=201)
async def create_user(body: CreateUserReq,                          # JSON 请求体 → Pydantictrace_id: str = Body("", embed=True),     # 单独的 body 字段x_real_ip: str | None = Header(None),       # 请求头
):return {"created": body, "trace_id": trace_id, "real_ip": x_real_ip}# POST /users
# Content-Type: application/json
# X-Real-IP: 192.168.1.1
# {"name": "赵六", "email": "zhao@test.com", "age": 30}

4.4 表单与文件

Python
from fastapi import Form, File, UploadFile# 表单数据（非 JSON）
@app.post("/login")
async def login(username: str = Form(...),password: str = Form(..., min_length=8),
):return {"username": username}# 文件上传
@app.post("/upload")
async def upload_file(file: UploadFile = File(...),description: str = Form(""),
):contents = await file.read()return {"filename": file.filename,"content_type": file.content_type,"size": len(contents),"description": description,}# 多文件上传
@app.post("/upload-multiple")
async def upload_multiple(files: List[UploadFile] = File(...),
):return [{"name": f.filename, "size": len(await f.read())} for f in files]

4.5 Cookie 与 Header

Python
from fastapi import Cookie, Header@app.get("/items")
async def read_items(session_id: str | None = Cookie(None),    # 从 Cookie 读取user_agent: str | None = Header(None),     # 从请求头读取x_token: str = Header(..., alias="X-Token"), # 自定义头部
):return {"session": session_id, "ua": user_agent, "token": x_token}

5. 响应模型与序列化

FastAPI 用 response_model 控制返回什么——自动过滤、序列化、生成文档。

5.1 基础用法

Python
from datetime import datetime
from typing import Listclass UserOut(BaseModel):    # 输出模型（不包含密码等敏感字段）model_config = ConfigDict(from_attributes=True)id: intname: stremail: strcreated_at: datetimeclass UserInDB(BaseModel):   # 数据库模型（含密码哈希）model_config = ConfigDict(from_attributes=True)id: intname: stremail: strhashed_password: strcreated_at: datetime@app.get("/users/{user_id}", response_model=UserOut)
async def get_user(user_id: int):# 假设从 DB 取出 UserInDB 对象db_user = UserInDB(id=user_id,name="测试",email="test@test.com",hashed_password="$2b$12$...",created_at=datetime.now(),)# response_model=UserOut 会自动过滤掉 hashed_passwordreturn db_user# 列表响应
@app.get("/users", response_model=List[UserOut])
async def list_users():# 返回列表自动处理return [db_user_1, db_user_2]

5.2 响应状态码与头

Python
from fastapi import status, Response@app.post("/items", status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemReq):return item# 手动设置响应头
@app.get("/custom-header")
async def custom(response: Response):response.headers["X-Custom-Header"] = "custom-value"response.set_cookie(key="session", value="abc123", httponly=True)return {"message": "ok"}

5.3 Union 响应与 discriminated 类型

Python
from typing import Unionclass SuccessResp(BaseModel):status: str = "ok"data: dictclass ErrorResp(BaseModel):status: str = "error"message: str@app.get("/result/{flag}", response_model=Union[SuccessResp, ErrorResp])
async def get_result(flag: bool):if flag:return SuccessResp(data={"key": "value"})return ErrorResp(message="发生了错误")

6. 依赖注入系统

FastAPI 的依赖注入（Dependency Injection, DI）是框架最优雅的设计之一——你声明"我需要什么"，框架自动提供。

6.1 基础依赖

Python
from fastapi import Depends, FastAPI# 定义依赖（就是一个函数）
async def get_db_session():"""模拟数据库连接"""  db = "数据库连接对象"try:yield db              # yield 之前的代码在请求开始时执行finally:print("关闭数据库连接")  # yield 之后的代码在响应返回后执行def get_current_user(token: str = Header(...)):if token != "secret":raise HTTPException(status_code=401)return {"username": "admin", "role": "admin"}@app.get("/protected")
async def protected_route(user: dict = Depends(get_current_user),db: str = Depends(get_db_session),
):return {"user": user, "db": db}# Depends(get_current_user) 自动提取 Header token → 校验 → 注入

6.2 可复用依赖与类依赖

Python
from dataclasses import dataclass# 类形式依赖（可接受参数）
class Pagination:def __init__(self, page: int = Query(1, ge=1),size: int = Query(10, ge=1, le=100)):self.page = pageself.size = sizeself.offset = (page - 1) * size@app.get("/items")
async def list_items(pagination: Pagination = Depends()):  # Depends() 自动注入return {"page": pagination.page,"offset": pagination.offset,"size": pagination.size,"items": []}# 多个路由共用分页依赖
@app.get("/users")
async def list_users(pg: Pagination = Depends()):return {"page": pg.page, "users": []}

6.3 全局依赖

Python
from fastapi import Depends# 应用到整个应用
app = FastAPI(dependencies=[Depends(verify_api_key)])# 或应用到路由组（APIRouter）
router = APIRouter(prefix="/admin",dependencies=[Depends(require_admin)],
)# 或应用到单个路由
@router.get("/dashboard")
async def dashboard(user: User = Depends(require_admin)):return {"dashboard": "data"}

Depends 的核心优势： 
复用性 — 分页、认证、DB 会话一次定义，到处注入 
可测试性 — 测试时可覆盖任何依赖 
类型安全 — 注入的类型即文档 
生命周期管理 — yield generator 自动处理资源释放

7. 错误处理与校验异常

7.1 异常处理器

Python
from fastapi import HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exception_handlers import http_exception_handler# 自定义 404 错误
@app.exception_handler(HTTPException)
async def custom_http_handler(request: Request, exc: HTTPException):return JSONResponse(status_code=exc.status_code,content={"code": exc.status_code,"message": exc.detail,"path": request.url.path,},headers=exc.headers,)# 处理未捕获异常
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):return JSONResponse(status_code=500,content={"detail": "服务器内部错误"},)# 路由中抛出 HTTP 异常
@app.get("/items/{item_id}")
async def read_item(item_id: int):if item_id == 0:raise HTTPException(status_code=404,detail="商品不存在",headers={"X-Error-Code": "NOT_FOUND"},)return {"item_id": item_id}

7.2 自定义校验错误

Python
from pydantic import ValidationError
from fastapi.exceptions import RequestValidationError@app.exception_handler(RequestValidationError)
async def validation_handler(request: Request, exc: RequestValidationError):errors = []for err in exc.errors():errors.append({"field": ".".join(str(loc) for loc in err["loc"]),"message": err["msg"],"type": err["type"],})return JSONResponse(status_code=422,content={"code": 422, "message": "参数校验失败", "errors": errors},)

7.3 业务异常体系

Python
# 定义业务异常类
class AppException(Exception):def __init__(self, code: int, message: str, status_code: int = 400):self.code = codeself.message = messageself.status_code = status_codeclass NotFoundError(AppException):def __init__(self, resource: str, resource_id):super().__init__(code=1001, message=f"{resource} {resource_id} 不存在", status_code=404)class PermissionDenied(AppException):def __init__(self):super().__init__(code=1003, message="权限不足", status_code=403)@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):return JSONResponse(status_code=exc.status_code,content={"code": exc.code, "message": exc.message},)# 路由中使用
@app.get("/users/{uid}")
async def get_user(uid: int):user = find_user(uid)if not user:raise NotFoundError("用户", uid)return user

8. 中间件与 CORS

8.1 CORS 跨域

Python
from fastapi.middleware.cors import CORSMiddlewareapp = FastAPI()app.add_middleware(CORSMiddleware,allow_origins=[                       # 允许的来源"http://localhost:5173",        # 前端 dev server"https://your-frontend.com",],allow_credentials=True,               # 允许 Cookie 携带allow_methods=["*"],                # 允许所有 HTTP 方法allow_headers=["*"],                # 允许所有请求头expose_headers=["X-Total-Count"],    # 暴露给前端的自定义头max_age=600,                        # 预检请求缓存时间（秒）
)# 生产环境不要用 allow_origins=["*"] + allow_credentials=True
# 这两个不能同时为通配符——spec 禁止

8.2 自定义中间件

Python
import time
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddlewareclass RequestTimingMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):start = time.time()response = await call_next(request)elapsed = time.time() - startresponse.headers["X-Process-Time"] = f"{elapsed:.4f}s"return response# 请求日志中间件
class AccessLogMiddleware(BaseHTTPMiddleware):async def dispatch(self, request: Request, call_next):print(f"→ {request.method} {request.url.path}")response = await call_next(request)print(f"← {response.status_code}")return responseapp.add_middleware(RequestTimingMiddleware)
app.add_middleware(AccessLogMiddleware)

8.3 路由级中间件（APIRouter）

Python
from fastapi import APIRouterrouter = APIRouter(prefix="/api/v1")@router.middleware("http")
async def router_middleware(request: Request, call_next):print(f"V1 路由: {request.url.path}")return await call_next(request)

9. 数据库集成 (SQLAlchemy + Pydantic)

生产中最常用的组合：SQLAlchemy 2.0 做 ORM，Pydantic 做请求/响应模型。两者通过 from_attributes=True 无缝桥接。

9.1 安装与配置

bash
$ pip install sqlalchemy asyncmy   # 异步 MySQL 驱动
$ pip install sqlalchemy aiosqlite  # 异步 SQLite（开发用）
$ pip install alembic               # 数据库迁移

Python
# database.py — 数据库配置
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine, async_sessionmaker
from sqlalchemy.orm import DeclarativeBase# 异步引擎 (MySQL 用 asyncmy, SQLite 用 aiosqlite)
DATABASE_URL = "sqlite+aiosqlite:///./app.db"
engine = create_async_engine(DATABASE_URL, echo=True)# 异步会话工厂
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)class Base(DeclarativeBase):pass# 依赖注入 DB 会话
async def get_db() -> AsyncSession:async with AsyncSessionLocal() as session:try:yield sessionawait session.commit()except Exception:await session.rollback()raise

9.2 定义模型（SQLAlchemy + Pydantic）

Python
# models.py — SQLAlchemy ORM 模型
from sqlalchemy import Column, Integer, String, DateTime, Boolean, Text, funcclass UserModel(Base):__tablename__ = "users"id = Column(Integer, primary_key=True, autoincrement=True)name = Column(String(50), nullable=False)email = Column(String(120), unique=True, nullable=False, index=True)age = Column(Integer, default=18)hashed_password = Column(String(255), nullable=False)is_active = Column(Boolean, default=True)created_at = Column(DateTime, server_default=func.now())updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

Python
# schemas.py — Pydantic 模型
from datetime import datetime
from typing import Optional# 请求模型
class UserCreate(BaseModel):name: str = Field(..., min_length=2, max_length=50)email: EmailStrage: int = Field(18, ge=0)password: str = Field(..., min_length=8)class UserUpdate(BaseModel):name: Optional[str] = Field(None, min_length=2, max_length=50)age: Optional[int] = Field(None, ge=0)# 响应模型 — from_attributes 允许从 ORM 对象创建
class UserResponse(BaseModel):model_config = ConfigDict(from_attributes=True)id: intname: stremail: strage: intis_active: boolcreated_at: Optional[datetime]# 翻页响应
class PaginatedUsers(BaseModel):items: List[UserResponse]total: intpage: intsize: inttotal_pages: int

9.3 CRUD 实战

Python
# crud.py — 数据库操作
from sqlalchemy import select, func
from sqlalchemy.ext.asyncio import AsyncSessionasync def create_user(db: AsyncSession, data: UserCreate) -> UserModel:user = UserModel(name=data.name,email=data.email,age=data.age,hashed_password="hashed_" + data.password,  # 实际用 passlib)db.add(user)await db.flush()          # 提交获取 idawait db.refresh(user)    # 刷新关联字段return userasync def get_user(db: AsyncSession, user_id: int) -> UserModel | None:result = await db.execute(select(UserModel).where(UserModel.id == user_id))return result.scalar_one_or_none()async def list_users(db: AsyncSession, page: int = 1, size: int = 10, keyword: str | None = None
) -> tuple[list[UserModel], int]:query = select(UserModel)if keyword:query = query.where(UserModel.name.contains(keyword))# 总数count_q = select(func.count()).select_from(query.subquery())total = (await db.execute(count_q)).scalar()# 分页query = query.offset((page - 1) * size).limit(size)result = await db.execute(query)users = result.scalars().all()return list(users), total

Python
# router.py — 路由层
router = APIRouter(prefix="/users", tags=["用户管理"])@router.post("", response_model=UserResponse, status_code=201)
async def create_user(data: UserCreate,db: AsyncSession = Depends(get_db),
):return await create_user(db, data)@router.get("/{user_id}", response_model=UserResponse)
async def get_user(user_id: int,db: AsyncSession = Depends(get_db),
):user = await get_user(db, user_id)if not user:raise NotFoundError("用户", user_id)return user@router.get("", response_model=PaginatedUsers)
async def list_users(page: int = Query(1, ge=1),size: int = Query(10, ge=1, le=100),keyword: str | None = None,db: AsyncSession = Depends(get_db),
):users, total = await list_users(db, page, size, keyword)return PaginatedUsers(items=users, total=total, page=page, size=size,total_pages=(total + size - 1) // size,)@router.patch("/{user_id}", response_model=UserResponse)
async def update_user(user_id: int,data: UserUpdate,db: AsyncSession = Depends(get_db),
):user = await get_user(db, user_id)if not user:raise NotFoundError("用户", user_id)if data.name is not None:user.name = data.nameif data.age is not None:user.age = data.ageawait db.flush()await db.refresh(user)return user

🗝 关键模式： 
• 三层分离：schemas.py（Pydantic 模型）→ crud.py（数据库操作）→ router.py（路由） 
• 异步 ORM：SQLAlchemy 2.0 的 AsyncSession 全程 async/await 
• 自动序列化：response_model=UserResponse + from_attributes=True 自动把 ORM 对象转 JSON 
• 依赖注入：DB 会话通过 Depends(get_db) 自动管理与清理

10. 认证与授权

10.1 JWT 认证（最常用）

bash
$ pip install python-jose[cryptography] passlib[bcrypt]

Python
# auth.py — 认证模块
from datetime import datetime, timedelta, timezone
from jose import JWTError, jwt
from passlib.context import CryptContextSECRET_KEY = "your-secret-key-change-in-production"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 60pwd_context = CryptContext(schemes=["bcrypt"])def verify_password(plain: str, hashed: str) -> bool:return pwd_context.verify(plain, hashed)def hash_password(plain: str) -> str:return pwd_context.hash(plain)def create_access_token(data: dict) -> str:to_encode = data.copy()expire = datetime.now(timezone.utc) + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)to_encode.update({"exp": expire})return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)def decode_access_token(token: str) -> dict:try:payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])return payloadexcept JWTError:raise HTTPException(status_code=401, detail="无效的 token")

Python
# auth_deps.py — 认证依赖
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestFormoauth2_scheme = OAuth2PasswordBearer(tokenUrl="/auth/login")async def get_current_user(token: str = Depends(oauth2_scheme),db: AsyncSession = Depends(get_db),
) -> UserModel:payload = decode_access_token(token)user_id = payload.get("sub")if not user_id:raise HTTPException(status_code=401)user = await get_user(db, int(user_id))if not user:raise HTTPException(status_code=401)return user# 角色检查
def require_role(role: str):async def role_checker(current_user: UserModel = Depends(get_current_user)):if current_user.role != role:raise HTTPException(status_code=403)return current_userreturn role_checker

Python
# router_auth.py
router = APIRouter(prefix="/auth", tags=["认证"])@router.post("/register", response_model=UserResponse)
async def register(data: UserCreate, db: AsyncSession = Depends(get_db)):user = UserModel(name=data.name, email=data.email, age=data.age,hashed_password=hash_password(data.password),)db.add(user); await db.flush(); await db.refresh(user)return user@router.post("/login")
async def login(form: OAuth2PasswordRequestForm = Depends(),db: AsyncSession = Depends(get_db),
):result = await db.execute(select(UserModel).where(UserModel.email == form.username))user = result.scalar_one_or_none()if not user or not verify_password(form.password, user.hashed_password):raise HTTPException(status_code=401, detail="邮箱或密码错误")token = create_access_token({"sub": str(user.id), "role": user.role})return {"access_token": token, "token_type": "bearer"}# 需要认证的路由
@router.get("/me", response_model=UserResponse)
async def get_me(current_user: UserModel = Depends(get_current_user)):return current_user# 需要管理员权限
@router.get("/admin-only")
async def admin_only(admin: UserModel = Depends(require_role("admin"))):return {"message": "欢迎管理员"}

11. 文件上传与下载

11.1 文件上传保存

Python
import os, uuid
from pathlib import PathUPLOAD_DIR = Path("uploads")
UPLOAD_DIR.mkdir(exist_ok=True)@app.post("/files/upload")
async def upload_file(file: UploadFile = File(...)):# 生成唯一文件名ext = Path(file.filename).suffix if file.filename else ".bin"unique_name = f"{uuid.uuid4()}{ext}"save_path = UPLOAD_DIR / unique_name# 分块写入（大文件友好）with open(save_path, "wb") as f:while chunk := await file.read(1024 * 1024):  # 1MB 块f.write(chunk)return {"filename": file.filename,"saved_as": unique_name,"size": save_path.stat().st_size,"url": f"/files/{unique_name}",}

11.2 文件下载

Python
from fastapi.responses import FileResponse, StreamingResponse# 方式一：FileResponse（直接下载）
@app.get("/files/{file_id}")
async def download_file(file_id: str):file_path = UPLOAD_DIR / file_idif not file_path.exists():raise HTTPException(404)return FileResponse(path=file_path,filename=file_id,                     # 下载时的文件名media_type="application/octet-stream",  # 强制下载)# 方式二：StreamingResponse（大文件/流式）
async def iter_file(path: Path, chunk_size=64 * 1024):  # 64KBwith open(path, "rb") as f:while chunk := f.read(chunk_size):yield chunk@app.get("/stream/{file_id}")
async def stream_file(file_id: str):file_path = UPLOAD_DIR / file_idreturn StreamingResponse(iter_file(file_path),media_type="application/octet-stream",headers={"Content-Disposition": f'attachment; filename="{file_id}"'},)

12. 后台任务与定时器

12.1 FastAPI 内置后台任务

Python
from fastapi import BackgroundTasksdef send_welcome_email(email: str):print(f"发送欢迎邮件到 {email}")def log_registration(user_id: int):print(f"记录注册日志: user_id={user_id}")@app.post("/users/register")
async def register(data: UserCreate,tasks: BackgroundTasks,
):user = await create_user(data)# 后台执行（不阻塞响应返回）tasks.add_task(send_welcome_email, user.email)tasks.add_task(log_registration, user.id)return user

12.2 定期任务（APScheduler）

bash
$ pip install apscheduler

Python
from contextlib import asynccontextmanager
from apscheduler.schedulers.asyncio import AsyncIOSchedulerscheduler = AsyncIOScheduler()async def cleanup_old_files():"""每天清理过期文件"""print("清理过期文件...")async def sync_external_data():"""每小时同步数据"""print("同步外部数据...")# 启动/关闭钩子（FastAPI 生命周期）
@asynccontextmanager
async def lifespan(app: FastAPI):# 启动时scheduler.add_job(cleanup_old_files, "cron", hour=3)       # 每天 3:00scheduler.add_job(sync_external_data, "interval", hours=1)  # 每小时scheduler.start()yield# 关闭时scheduler.shutdown()app = FastAPI(lifespan=lifespan)

13. WebSocket

Python
from fastapi import WebSocket, WebSocketDisconnectclass ConnectionManager:def __init__(self):self.active_connections: List[WebSocket] = []async def connect(self, ws: WebSocket):await ws.accept()self.active_connections.append(ws)def disconnect(self, ws: WebSocket):self.active_connections.remove(ws)async def broadcast(self, message: str):for connection in self.active_connections:await connection.send_text(message)manager = ConnectionManager()@app.websocket("/ws/{client_id}")
async def websocket_endpoint(ws: WebSocket, client_id: str):await manager.connect(ws)try:while True:data = await ws.receive_text()print(f"收到 {client_id}: {data}")await ws.send_text(f"服务器已收到: {data}")await manager.broadcast(f"[{client_id}] {data}")except WebSocketDisconnect:manager.disconnect(ws)await manager.broadcast(f"[{client_id}] 已断开连接")

💡 WebSocket vs HTTP： 
• HTTP = 请求-响应模式，每次新建连接（或复用 keep-alive），适合 CRUD 
• WebSocket = 全双工长连接，适合实时消息、聊天、通知、协同编辑 
• FastAPI 同一个应用可以同时提供 HTTP REST + WebSocket

14. 测试

FastAPI 基于 Starlette 的 TestClient，基于 httpx（同步/异步），模拟 HTTP 请求无需启动服务器。

bash
$ pip install pytest httpx

Python
# test_main.py
import pytest
from httpx import ASGITransport
from fastapi.testclient import TestClientfrom main import appclient = TestClient(app)def test_health_check():response = client.get("/ping")assert response.status_code == 200assert response.json() == {"status": "ok"}def test_create_user():payload = {"name": "测试用户","email": "test@example.com","age": 25,"password": "Test1234",}response = client.post("/users", json=payload)assert response.status_code == 201data = response.json()assert data["name"] == "测试用户"assert "id" in dataassert "password" not in data  # 响应模型过滤了敏感字段def test_validation_error():# 测试校验失败response = client.post("/users", json={"name": "x"})  # 太短 + 缺少必填字段assert response.status_code == 422errors = response.json()assert "detail" in errorsdef test_auth_fail():response = client.get("/auth/me")  # 未传 tokenassert response.status_code == 401

异步测试

Python
import pytest
from httpx import AsyncClient, ASGITransport@pytest.mark.asyncio
async def test_async_endpoint():transport = ASGITransport(app=app)async with AsyncClient(transport=transport, base_url="http://test") as ac:response = await ac.get("/ping")assert response.status_code == 200

覆盖依赖（测试关键技巧）

Python
# 测试时替换数据库
from main import app, get_dbasync def override_get_db():async with TestSessionLocal() as session:yield sessionapp.dependency_overrides[get_db] = override_get_db# 之后的 TestClient 请求就会使用测试数据库
client = TestClient(app)

15. 部署与生产配置

15.1 使用 Gunicorn + Uvicorn Worker

bash
$ pip install gunicorn# 多进程部署（推荐）
$ gunicorn main:app \--worker-class uvicorn.workers.UvicornWorker \--workers 4 \--bind 0.0.0.0:8000 \--timeout 120 \--access-logfile - \--error-logfile -

15.2 Docker 部署

Dockerfile
FROM python:3.12-slimWORKDIR /app# 依赖安装
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt# 应用代码
COPY . .# 健康检查（K8s 需要）
HEALTHCHECK --interval=30s --timeout=3s \CMD python -c "import http.client; http.client.HTTPConnection('localhost', 8000).request('GET', '/ping')"CMD ["gunicorn", "main:app", \"--worker-class", "uvicorn.workers.UvicornWorker", \"--workers", "4", \"--bind", "0.0.0.0:8000"]

.env
# 环境变量配置
DATABASE_URL=postgresql+asyncpg://user:pass@localhost:5432/db
SECRET_KEY=your-256-bit-secret
ACCESS_TOKEN_EXPIRE_MINUTES=60
LOG_LEVEL=info
CORS_ORIGINS=https://frontend.example.com

Python
# config.py — 配置管理（Pydantic Settings）
from pydantic_settings import BaseSettingsclass Settings(BaseSettings):app_name: str = "My API"debug: bool = Falsedatabase_url: strsecret_key: straccess_token_expire_minutes: int = 60cors_origins: str = "*"log_level: str = "info"model_config = ConfigDict(env_file=".env", extra="ignore")settings = Settings()  # 自动从 .env / 环境变量读取# 在 app 中使用
app = FastAPI(title=settings.app_name, debug=settings.debug)

15.3 生产 Checklist

16. 综合实战：完整 RESTful API

一个完整的博客系统 API，串起所有知识点。目录结构如下：

结构
myblog/
├── main.py               # 应用入口
├── config.py             # 配置管理
├── database.py           # 数据库引擎 + session
├── models/               # SQLAlchemy ORM 模型
│   ├── __init__.py
│   ├── user.py
│   └── post.py
├── schemas/              # Pydantic 模型
│   ├── __init__.py
│   ├── user.py
│   └── post.py
├── crud/                 # 数据库操作
│   ├── __init__.py
│   ├── user.py
│   └── post.py
├── routers/              # API 路由
│   ├── __init__.py
│   ├── auth.py
│   ├── users.py
│   └── posts.py
├── deps.py               # 公共依赖
├── exceptions.py         # 业务异常
└── tests/                # 测试

Python
# main.py — 完整应用入口
import asyncio
import time
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse# 业务异常处理器
from exceptions import AppException# 路由
from routers.auth import router as auth_router
from routers.users import router as user_router
from routers.posts import router as post_router# 数据库初始化
from database import engine, Base@asynccontextmanager
async def lifespan(app: FastAPI):# 启动时创建表async with engine.begin() as conn:await conn.run_sync(Base.metadata.create_all)yield# 关闭引擎await engine.dispose()app = FastAPI(title="Blog API",description="FastAPI + SQLAlchemy + Pydantic 综合实战",version="1.0.0",lifespan=lifespan,
)# 中间件
app.add_middleware(CORSMiddleware,allow_origins=["http://localhost:3000"],allow_credentials=True,allow_methods=["*"],allow_headers=["*"],
)# 请求耗时中间件
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):start = time.time()response = await call_next(request)response.headers["X-Process-Time"] = f"{time.time() - start:.4f}"return response# 异常处理器
@app.exception_handler(AppException)
async def app_exception_handler(request: Request, exc: AppException):return JSONResponse(status_code=exc.status_code,content={"code": exc.code, "message": exc.message},)# 注册路由
app.include_router(auth_router, prefix="/api/v1")
app.include_router(user_router, prefix="/api/v1")
app.include_router(post_router, prefix="/api/v1")@app.get("/")
async def root():return {"name": "Blog API", "version": "1.0.0", "docs": "/docs"}if __name__ == "__main__":import uvicornuvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

Python
# schemas/post.py — 文章 Pydantic 模型
from pydantic import BaseModel, ConfigDict, Field
from datetime import datetime
from typing import List, Optionalclass PostCreate(BaseModel):title: str = Field(..., min_length=2, max_length=200)content: str = Field(..., min_length=10)published: bool = Truetags: List[str] = []class PostUpdate(BaseModel):title: Optional[str] = Field(None, min_length=2, max_length=200)content: Optional[str] = Field(None, min_length=10)published: Optional[bool] = Nonetags: Optional[List[str]] = Noneclass PostOut(BaseModel):model_config = ConfigDict(from_attributes=True)id: inttitle: strcontent: strpublished: booltags: str                 # SQLite 用逗号分隔存储author_id: intauthor_name: str            # 关联用户created_at: datetimeupdated_at: Optional[datetime]

Python
# routers/posts.py — 文章 CRUD 路由
router = APIRouter(prefix="/posts", tags=["文章管理"])@router.post("", response_model=PostOut, status_code=201)
async def create_post(data: PostCreate,current_user: UserModel = Depends(get_current_user),db: AsyncSession = Depends(get_db),
):post = PostModel(title=data.title,content=data.content,published=data.published,tags=",".join(data.tags),author_id=current_user.id,)db.add(post)await db.flush()await db.refresh(post)return await _format_post(db, post)@router.get("", response_model=List[PostOut])
async def list_posts(page: int = Query(1, ge=1),size: int = Query(10, ge=1, le=50),tag: str | None = None,db: AsyncSession = Depends(get_db),
):query = select(PostModel).where(PostModel.published == True)if tag:query = query.where(PostModel.tags.contains(tag))query = query.offset((page - 1) * size).limit(size).order_by(PostModel.created_at.desc())result = await db.execute(query)posts = result.scalars().all()return [await _format_post(db, p) for p in posts]@router.get("/{post_id}", response_model=PostOut)
async def get_post(post_id: int, db: AsyncSession = Depends(get_db)):result = await db.execute(select(PostModel).where(PostModel.id == post_id))post = result.scalar_one_or_none()if not post:raise NotFoundError("文章", post_id)return await _format_post(db, post)@router.delete("/{post_id}", status_code=204)
async def delete_post(post_id: int,current_user: UserModel = Depends(get_current_user),db: AsyncSession = Depends(get_db),
):result = await db.execute(select(PostModel).where(PostModel.id == post_id))post = result.scalar_one_or_none()if not post:raise NotFoundError("文章", post_id)if post.author_id != current_user.id and current_user.role != "admin":raise PermissionDenied()await db.delete(post)async def _format_post(db, post):"""格式化文章输出（获取作者名字）"""user = await db.get(UserModel, post.author_id)return {"id": post.id,"title": post.title,"content": post.content,"published": post.published,"tags": post.tags,"author_id": post.author_id,"author_name": user.name if user else "未知","created_at": post.created_at,"updated_at": post.updated_at,}

综合实战的模式： 
• 三层架构：路由（关注 HTTP）+ CRUD（关注 DB）+ Schema（关注数据形状） 
• 依赖注入贯穿始终：get_db（会话管理）+ get_current_user（认证）+ 业务参数 
• Pydantic 双向绑定：请求校验（PostCreate）→ 业务处理 → 响应序列化（PostOut） 
• 错误处理体系：业务异常（NotFoundError / PermissionDenied）→ 统一全局 Handler 
• 异步全链路：async DB → async route → async middleware，不阻塞事件循环