虚拟环境

从 Python 3.3 开始，venv 就是标准库的一部分。
你不需要安装任何第三方工具，只要有 Python 就能创建虚拟环境。

venv 做的事情很纯粹——创建一个独立的 Python 环境，复制必要的二进制文件和库。它不做任何"魔法"，创建出来的环境结构清晰，你可以随时查看 .venv 目录里到底有什么。

在 Windows、macOS、Linux 上，venv 的行为和命令完全一致，只是激活脚本的路径略有不同（Windows 是 .venv\Scripts\activate，Unix 系是 .venv/bin/activate）。这让跨平台开发变得简单。

✅ venv + requirements.txt 这是 Python 社区最经典、最成熟的依赖管理方式：

# 创建虚拟环境
python -m venv .venv

# 生成 【依赖列表】
pip freeze > requirements.txt

# 根据 【依赖列表】安装
pip install -r requirements.txt

# 根据 【依赖列表】安装，使用国内镜像源
pip install -r requirements.txt -i https://pypi.tuna.tsinghua.edu.cn/simple

# 安装三方库
pip install 三方库1 三方库2 -i https://pypi.tuna.tsinghua.edu.cn/simple

前言

FastAPI 是基于 Python 类型提示构建的高性能 Web 框架，原生支持异步、自动生成接口文档、内置参数校验与数据序列化能力，是当前 Python 后端生态中工程化效率最高的框架之一。
本文从基础环境搭建到完整项目实战，覆盖路由管理、参数解析、数据校验、异常处理、统一响应等核心工程能力，所有代码均可直接复制运行。

---

一、安装

1.1 环境安装

FastAPI 运行依赖 ASGI 服务器，官方推荐使用 uvicorn。执行以下命令安装核心依赖：

pip install fastapi uvicorn

1.2 最小可运行示例

创建 main.py 文件，编写最简 FastAPI 应用：

# main.py
from fastapi import FastAPI

# 创建 FastAPI 应用实例，可配置标题、描述等文档信息
wtt_app = FastAPI(
    title="FastAPI 入门示例",
    description="第一个 FastAPI 程序，演示基础能力",
    version="1.0.0"
)

# 装饰器注册 GET 路由，路径为根路径
@wtt_app.get("/")
def root():
    return {"message": "Hello FastAPI!"}

1.3 服务启动方式

在终端执行以下命令启动服务，--reload 参数表示代码修改后自动重启，仅开发环境使用：

uvicorn main:wtt_app --reload

启动成功后访问 http://127.0.0.1:8000 即可看到返回结果。

1.4 自动生成接口文档

FastAPI 基于 OpenAPI 规范自动生成两份接口文档，启动服务后直接访问：

• Swagger UI 交互式文档：http://127.0.0.1:8000/docs（可在线调试接口）
• ReDoc 文档：http://127.0.0.1:8000/redoc（适合阅读的静态文档）

---

二、路由路径注册与模块化管理

2.1 基础路由与路径参数

通过路径中 {变量名} 的方式定义路径参数，配合 Python 类型注解可自动完成类型转换与校验：

# 追加到 main.py 中
@app.get("/items/{item_id}", summary="获取商品详情")
def get_item(item_id: int):
    """
    - item_id: 路径参数, 用于 用 【 :int 】 进行修饰，所以 必须为整数，FastAPI 会自动做类型校验
    """
    return {"item_id": item_id, "name": f"商品{item_id}"}

测试命令：

curl http://127.0.0.1:8000/items/100

若传入非数字参数，FastAPI 会自动返回 422 参数校验错误。

2.2 路由层级与请求方法

FastAPI 支持所有标准 HTTP 方法（GET/POST/PUT/DELETE 等），同一路径可绑定不同请求方法：

# 追加到 main.py 中
@app.post("/items", summary="创建商品")
def create_item():
    return {"message": "商品创建成功"}

@app.put("/items/{item_id}", summary="更新商品")
def update_item(item_id: int):
    return {"item_id": item_id, "message": "商品更新成功"}

2.3 APIRouter 路由组 管理

大型项目中不能将所有路由写在同一个文件中，FastAPI 提供 APIRouter 实现路由模块化拆分，支持路由前缀、标签统一配置。

第一步：创建用户路由模块 routers/user.py

# routers/user.py
from fastapi import APIRouter

# 创建路由实例
# prefix: 路由统一前缀，该模块下所有路由自动添加 /user 前缀
# tags: 接口文档分组标签
user_router = APIRouter(prefix="/user", tags=["用户管理"])

@user_router.get("/{user_id}", summary="获取用户信息")
def get_user(user_id: int):
    return {"user_id": user_id, "username": f"用户{user_id}"}

@user_router.post("/add", summary="新增用户")
def add_user():
    return {"message": "用户创建成功"}

第二步：主应用挂载路由

# main.py 中追加导入与挂载
from routers.user import user_router

# 挂载用户路由模块
app.include_router(user_router)

测试命令：

curl http://127.0.0.1:8000/user/1

访问文档 /docs 可看到接口已按「用户管理」分组展示。

---

三、路由函数参数解析（GET/POST 全场景）

3.1 GET 请求参数解析

GET 请求参数分为两类：路径参数（嵌入 URL）、查询参数（URL 中 ? 后面的键值对）。

3.1.1 路径参数 + 查询参数组合

# 追加到 routers/user.py 中
@user_router.get("/list", summary="获取用户列表")
def get_user_list(
    page: int = 1,    # 查询参数，带默认值
    page_size: int = 10
):
    """
    page、page_size 均为查询参数，未传时使用默认值
    """
    return {
        "page": page,
        "page_size": page_size,
        "list": [{"id": i, "name": f"用户{i}"} for i in range(page_size)]
    }

测试命令：

# 不传参，使用默认值
curl http://127.0.0.1:8000/user/list

# 传参覆盖默认值
curl "http://127.0.0.1:8000/user/list?page=2&page_size=5"

3.1.2 可选查询参数

通过 Optional 或 类型 | None 声明可选参数，未传时值为 None：

from typing import Optional

@user_router.get("/search", summary="搜索用户")
def search_user(
    keyword: str,
    status: Optional[int] = None  # 可选参数
):
    return {"keyword": keyword, "status": status}

3.2 POST 请求参数解析

3.2.1 基础请求体（Pydantic 模型）

POST 请求体通过 Pydantic 的 BaseModel 定义，FastAPI 自动解析 JSON 请求体并做类型校验。

from pydantic import BaseModel

# 定义请求体模型
class UserCreate(BaseModel):
    username: str
    age: int
    email: str | None = None  # 可选字段, 语法解释 { : str | None 是 类型注解 } 和 { = None 默认值 } 

@user_router.post("/create", summary="创建用户（请求体方式）")
def create_user(user: UserCreate):
    # 直接通过属性访问参数
    return {
        "username": user.username,
        "age": user.age,
        "email": user.email
    }

测试命令：

curl -X POST http://127.0.0.1:8000/user/create \
  -H "Content-Type: application/json" \
  -d '{"username":"zhangsan","age":20,"email":"zhangsan@example.com"}'

3.2.2 嵌套请求模型 （类 的 组合模式）

Pydantic 支持模型嵌套，适合复杂结构的请求体：

class Address(BaseModel):
    province: str
    city: str
    detail: str

class UserWithAddress(BaseModel):
    username: str
    age: int
    address: Address  # 嵌套模型

@user_router.post("/create_with_address", summary="创建用户（带地址）")
def create_user_with_address(user: UserWithAddress):
    return {
        "username": user.username,
        "address": user.address.city
    }

3.2.3 表单数据解析

解析 application/x-www-form-urlencoded 格式的表单数据，需先安装依赖：

pip install python-multipart

代码示例：

from fastapi import Form

@user_router.post("/login", summary="表单登录")
def login(
    username: str = Form(..., description="用户名"),
    password: str = Form(..., description="密码")
):
    return {"username": username, "login_status": True}

测试命令：

curl -X POST http://127.0.0.1:8000/user/login \
  -d "username=admin&password=123456"

3.2.4 文件上传接收

FastAPI 支持单文件、多文件上传，通过 UploadFile 接收文件对象：

from fastapi import File, UploadFile

@user_router.post("/upload_avatar", summary="上传头像")
async def upload_avatar(file: UploadFile = File(..., description="头像文件")):
    # 读取文件内容（异步读取）
    content = await file.read()
    return {
        "filename": file.filename,
        "content_type": file.content_type,
        "file_size": len(content)
    }

测试命令：

curl -X POST http://127.0.0.1:8000/user/upload_avatar \
  -F "file=@avatar.png"

---

四、前端参数校验：原生能力全解

FastAPI 结合 Pydantic 实现零额外依赖的参数校验，覆盖基础约束到自定义规则全场景。

4.1 基础校验规则

4.1.1 路径/查询参数校验

通过 Path、Query 类配置校验规则，支持数值范围、字符串长度等约束：

from fastapi import Path, Query

@user_router.get("/check/{user_id}", summary="参数校验示例")
def check_param(
    # 路径参数校验：必须大于0
    user_id: int = Path(..., gt=0, description="用户ID，必须大于0"),
    # 查询参数校验：字符串长度3-20
    username: str = Query(..., min_length=3, max_length=20, description="用户名")
):
    return {"user_id": user_id, "username": username}

4.1.2 请求体模型校验

通过 Field 为模型字段配置校验规则，是业务开发中最常用的方式：

from pydantic import BaseModel, Field

class UserCreateV2(BaseModel):
    username: str = Field(min_length=3, max_length=20, description="用户名，3-20字符")
    age: int = Field(ge=0, le=120, default=18, description="年龄，0-120岁")
    phone: str = Field(pattern=r"^1[3-9]\d{9}$", description="手机号格式校验")
    email: str | None = Field(default=None, description="可选邮箱")

4.1.3 枚举值校验

使用 Python 标准库 Enum 限制参数只能为指定枚举值：

from enum import Enum

class UserStatus(str, Enum):
    NORMAL = "normal"
    DISABLED = "disabled"
    VIP = "vip"

@user_router.get("/status/{status}", summary="枚举值校验")
def get_user_by_status(status: UserStatus):
    return {"status": status.value, "message": f"查询{status.value}状态用户"}

4.2 进阶校验能力

4.2.1 自定义字段校验函数

通过 Pydantic 的 @field_validator 装饰器实现复杂自定义校验逻辑：

from pydantic import field_validator

class UserRegister(BaseModel):
    username: str = Field(min_length=3, max_length=20)
    password: str = Field(min_length=6, max_length=32)

    @field_validator("password")
    @classmethod
    def validate_password(cls, value: str) -> str:
        # 自定义规则：密码必须包含至少一位数字
        if not any(char.isdigit() for char in value):
            raise ValueError("密码必须包含至少一位数字")
        return value

校验失败时会自动返回 422 错误，错误信息包含自定义提示。

4.2.2 校验失败默认响应格式

参数校验失败时，FastAPI 默认返回如下结构的 422 响应：

{
  "detail": [
    {
      "loc": ["body", "password"],
      "msg": "密码必须包含至少一位数字",
      "type": "value_error"
    }
  ]
}

后续全局异常处理章节会讲解如何将其封装为统一业务格式。

---

五、全局异常处理：统一错误响应

5.1 内置 HTTPException 使用

业务逻辑中可直接抛出 HTTPException 终止请求，返回指定状态码和错误信息：

from fastapi import HTTPException

@user_router.get("/{user_id}/info", summary="查询用户（演示异常抛出）")
def get_user_info(user_id: int):
    if user_id <= 0:
        # 抛出HTTP异常，指定状态码和错误详情
        raise HTTPException(status_code=400, detail="用户ID不合法")
    if user_id > 1000:
        raise HTTPException(status_code=404, detail="用户不存在")
    return {"user_id": user_id, "username": "test_user"}

5.2 全局异常处理器注册

通过 @app.exception_handler() 装饰器注册异常处理器，可拦截指定类型的异常并自定义返回格式。

5.3 自定义业务异常类

实际项目中通常定义业务异常，区分系统错误与业务错误：

# common/exceptions.py
class BizException(Exception):
    """自定义业务异常类"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(message)

5.4 常见异常统一封装

在主应用中注册全局异常处理器，覆盖业务异常、参数校验异常、404、500 等场景：

# main.py 追加内容
from fastapi import Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from common.exceptions import BizException

# 1. 捕获自定义业务异常
@app.exception_handler(BizException)
async def biz_exception_handler(request: Request, exc: BizException):
    return JSONResponse(
        status_code=200,  # 业务异常HTTP状态码统一200，通过业务code区分
        content={
            "code": exc.code,
            "message": exc.message,
            "data": None
        }
    )

# 2. 捕获参数校验异常
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    # 提取第一个错误信息
    first_error = exc.errors()[0]
    error_msg = f"参数[{'.'.join(map(str, first_error['loc']))}]：{first_error['msg']}"
    return JSONResponse(
        status_code=400,
        content={
            "code": 400,
            "message": error_msg,
            "data": None
        }
    )

# 3. 捕获HTTP异常（404、405等）
@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={
            "code": exc.status_code,
            "message": exc.detail,
            "data": None
        }
    )

# 4. 捕获服务器内部异常（500）
@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    # 生产环境可在此处记录错误日志
    return JSONResponse(
        status_code=500,
        content={
            "code": 500,
            "message": "服务器内部错误",
            "data": None
        }
    )

业务异常使用示例：

@user_router.get("/biz_test/{user_id}", summary="演示业务异常")
def biz_test(user_id: int):
    if user_id == 0:
        # 抛出自定义业务异常
        raise BizException(code=1001, message="用户ID不能为0")
    return {"user_id": user_id}

---

六、统一数据格式返回：标准化响应

6.1 通用响应结构设计

前后端分离项目通常采用 code + message + data 三段式统一响应结构，便于前端统一处理。

6.2 统一响应封装工具

创建通用响应模型与工具函数，所有接口统一返回该结构：

# common/response.py
from pydantic import BaseModel
from typing import Generic, TypeVar, Optional

# 泛型，支持不同类型的data字段
T = TypeVar("T")

class ResponseResult(BaseModel, Generic[T]):
    """统一响应模型"""
    code: int = 200
    message: str = "success"
    data: Optional[T] = None

def success(data: T = None, message: str = "操作成功") -> ResponseResult[T]:
    """成功响应封装"""
    return ResponseResult(code=200, message=message, data=data)

def fail(code: int = 400, message: str = "操作失败") -> ResponseResult:
    """失败响应封装"""
    return ResponseResult(code=code, message=message, data=None)

6.3 Response Model 响应模型

通过 response_model 参数指定接口返回模型，可自动过滤多余字段，保证数据安全：

from common.response import ResponseResult, success

# 定义用户响应模型（不返回敏感字段）
class UserResp(BaseModel):
    user_id: int
    username: str
    email: str | None

@user_router.get("/resp/{user_id}", summary="统一响应格式示例", response_model=ResponseResult[UserResp])
def get_user_resp(user_id: int):
    # 模拟数据库返回的完整数据（含密码等敏感字段）
    user_data = {
        "user_id": user_id,
        "username": "zhangsan",
        "password": "123456",  # 该字段会被自动过滤
        "email": "zhangsan@example.com"
    }
    return success(data=user_data)

接口返回结果中不会包含 password 字段，保证数据安全。

---

七、中间件：请求全链路全局管控

中间件（Middleware）是 FastAPI 中拦截 HTTP 请求与响应的全局钩子，运行在「客户端请求」和「路由函数」之间，遵循经典的洋葱模型：请求进来时逐层穿透中间件，响应返回时再逐层反向穿透。

它适合处理所有接口都需要的通用逻辑，比如：请求日志、跨域处理、接口耗时统计、全局鉴权、响应头统一添加等，避免在每个路由函数中重复编写代码。

7.1 中间件基础语法

FastAPI 提供两种使用中间件的方式：自定义中间件（用装饰器定义）和内置/第三方中间件（用 add_middleware 注册）。

7.1.1 自定义中间件标准结构

使用 @app.middleware("http") 装饰器定义全局 HTTP 中间件，固定结构如下：

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import time

app = FastAPI()

@app.middleware("http")
async def custom_middleware(request: Request, call_next):
    # ========== 请求阶段：路由执行前执行 ==========
    # 在这里做请求预处理：日志、鉴权、参数校验等
    start_time = time.time()

    # 调用 call_next 将请求传递给下一个中间件/路由函数
    # 它会返回最终的响应对象
    response = await call_next(request)

    # ========== 响应阶段：路由执行后执行 ==========
    # 在这里做响应后处理：添加响应头、统计耗时、日志收尾等
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)

    # 必须返回响应对象
    return response

核心说明：

• @app.middleware(“http”) 里的 "http" 目前是固定的唯一合法写法，不能随意替换为其他字符串。
• request: Request：封装了完整的请求信息（路径、方法、请求头、参数等）
• call_next：一个可调用对象，调用它才会将请求向下传递；不调用则请求直接终止
• 中间件必须是异步函数（async def），call_next 必须用 await 调用
• 最终必须返回响应对象

7.2 高频实战场景

7.2.1 场景一：接口耗时统计

这是最常用的中间件场景，自动为所有接口计算执行耗时并写入响应头。

from fastapi import FastAPI
import time
from fastapi import Request

# 创建 FastAPI 应用实例，可配置标题、描述等文档信息
app = FastAPI(
    title="FastAPI 入门示例",
    description="第一个 FastAPI 程序，演示基础能力",
    version="1.0.0"
)
@app.middleware("http")
async def calc_process_time(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start
    # 将耗时写入自定义响应头，单位秒
    print(f"此次请求耗时{duration:.4f}秒")
    response.headers["X-Process-Time"] = f"{duration:.4f}"
    return response


# 追加到 main.py 中
@app.get("/items/{item_id}", summary="获取商品详情")
def get_item(item_id: int):
    """
    - item_id: 路径参数，必须为整数，FastAPI 会自动做类型校验
    """
    return {"item_id": item_id, "name": f"商品{item_id}"}

验证方式：调用任意接口后，查看响应头即可看到 X-Process-Time 字段。

curl -I http://127.0.0.1:8000/items/112

7.2.2 场景二：全局请求日志

统一记录所有接口的请求方法、路径、状态码、耗时，替代每个接口手写日志。

import time
import logging
from fastapi import Request

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

@app.middleware("http")
async def access_log(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start

    # 记录日志：请求方法、路径、状态码、耗时
    logger.info(
        f"{request.method} {request.url.path} "
        f"- Status: {response.status_code} "
        f"- Cost: {duration:.4f}s"
    )
    return response

启动服务后访问任意接口，控制台会自动输出日志：

INFO:__main__:GET /health - Status: 200 - Cost: 0.0005s

7.2.3 场景三：简易全局鉴权

对需要登录的接口统一校验 Token，白名单路径（如文档、登录接口）直接放行。

from fastapi import Request
from fastapi.responses import JSONResponse

# 白名单：不需要鉴权的路径
WHITE_LIST = ["/docs", "/openapi.json", "/redoc", "/health", "/user/login"]

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    # 白名单路径直接放行
    if request.url.path in WHITE_LIST:
        return await call_next(request)

    # 从请求头获取 Token
    token = request.headers.get("Authorization")
    if not token or not token.startswith("Bearer "):
        return JSONResponse(
            status_code=401,
            content={"code": 401, "message": "缺少有效Token", "data": None}
        )

    # 简单校验（实际项目中替换为 JWT 解析逻辑）
    real_token = token.split(" ")[1]
    if real_token != "admin123":
        return JSONResponse(
            status_code=403,
            content={"code": 403, "message": "Token无效或已过期", "data": None}
        )

    # 将用户信息存入 request.state，路由函数中可直接取用
    request.state.user_id = 1
    request.state.username = "admin"

    return await call_next(request)

在路由函数中获取用户信息：

from fastapi import Request

@app.get("/user/info")
def get_current_user(request: Request):
    # 直接从 request.state 读取中间件注入的用户信息
    return {
        "user_id": request.state.user_id,
        "username": request.state.username
    }

测试命令：

# 不带 Token，返回 401
curl http://127.0.0.1:8000/user/info

# 带正确 Token，正常返回
curl -H "Authorization: Bearer admin123" http://127.0.0.1:8000/user/info

注意：简单鉴权可以用中间件实现，复杂的权限控制更推荐使用 FastAPI 的依赖注入（Depends），粒度更细、可复用性更强。

7.3 内置核心中间件：CORS 跨域处理

前后端分离项目中跨域是必解问题，FastAPI 官方内置了 CORSMiddleware，无需额外安装依赖，几行配置即可解决跨域问题。

7.3.1 完整配置示例

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware

app = FastAPI()

# 配置允许跨域的源（前端地址）
origins = [
    "http://localhost:3000",    # 前端开发地址
    "http://localhost:8080",
    "https://your-production.com",  # 生产环境前端域名
]

# 注册 CORS 中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,      # 允许的源列表
    allow_credentials=True,     # 允许携带 Cookie、Authorization 等凭证
    allow_methods=["*"],        # 允许所有 HTTP 方法（GET/POST/PUT/DELETE 等）
    allow_headers=["*"],        # 允许所有请求头
)

7.3.2 关键参数说明

参数	作用	注意事项
allow_origins	允许跨域访问的前端域名列表	生产环境禁止使用 ["*"]，存在安全风险
allow_credentials	是否允许携带凭证（Cookie、Token）	设为 True 时，allow_origins 不能为 *
allow_methods	允许的 HTTP 请求方法	常用 ["*"] 或 ["GET", "POST", "PUT", "DELETE"]
allow_headers	允许的请求头字段	常用 ["*"] 放行所有请求头

开发环境可以临时使用 allow_origins=["*"] 快速调试，上线前必须替换为明确的域名列表。

7.4 多个中间件的执行顺序（高频易错点）

当注册多个中间件时，执行顺序遵循洋葱模型：

• 请求阶段：按注册的逆序执行（最后注册的最先执行）
• 响应阶段：按注册的正序执行（最先注册的最先执行）

示例验证

from fastapi import FastAPI, Request

app = FastAPI()

@app.middleware("http")
async def middleware_a(request: Request, call_next):
    print("A 请求阶段")
    response = await call_next(request)
    print("A 响应阶段")
    return response

@app.middleware("http")
async def middleware_b(request: Request, call_next):
    print("B 请求阶段")
    response = await call_next(request)
    print("B 响应阶段")
    return response

@app.get("/")
def root():
    print("路由函数执行")
    return {"message": "ok"}

控制台输出顺序：

B 请求阶段
A 请求阶段
路由函数执行
A 响应阶段
B 响应阶段

可视化执行流程：

请求 → B → A → 路由函数 → A → B → 响应

工程化最佳实践：注册顺序建议从外到内：CORS中间件 → 日志中间件 → 鉴权中间件 → 业务中间件。

7.5 集成到实战项目

将中间件能力整合到前文的实战项目中，修改 app/main.py 如下：

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware
import time
import logging

from app.routers.user import user_router
from app.routers.upload import upload_router
from app.common.exceptions import BizException

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

app = FastAPI(
    title="FastAPI 实战项目",
    description="整合路由、校验、异常、中间件、统一响应的完整示例",
    version="1.0.0"
)

# ========== 1. CORS 跨域中间件（最外层，最先注册） ==========
origins = [
    "http://localhost:3000",
    "http://localhost:8080",
]
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ========== 2. 全局日志 + 耗时统计中间件 ==========
@app.middleware("http")
async def access_log_middleware(request: Request, call_next):
    start = time.perf_counter()
    response = await call_next(request)
    duration = time.perf_counter() - start

    response.headers["X-Process-Time"] = f"{duration:.4f}"
    logger.info(
        f"{request.method} {request.url.path} "
        f"- Status: {response.status_code} "
        f"- Cost: {duration:.4f}s"
    )
    return response

# ========== 挂载路由 ==========
app.include_router(user_router)
app.include_router(upload_router)

# ========== 全局异常处理（同前文，此处省略重复代码） ==========
# ... 保留原有的异常处理器代码 ...

---

补充说明

中间件是全局生效的，适合处理全接口通用逻辑；如果只需要部分接口复用逻辑，更推荐使用 FastAPI 的依赖注入（Depends），粒度更灵活、测试性更好。

需要我继续补充依赖注入（Depends） 章节，或者讲解 JWT 认证、数据库集成等进阶内容吗？

八、完整实战项目：整合所有能力

本章节将路由管理、参数校验、全局异常、统一响应、中间件五大核心能力整合为一个可直接运行的工程化项目，采用模块化分层结构，符合企业级开发规范。

8.1 项目目录结构

fastapi_demo/
├── app/
│   ├── __init__.py
│   ├── main.py              # 项目入口：注册中间件、异常处理器、挂载路由
│   ├── common/              # 公共组件层
│   │   ├── __init__.py
│   │   ├── response.py      # 统一响应结构封装
│   │   └── exceptions.py    # 自定义业务异常类
│   ├── middleware/          # 中间件层（新增）
│   │   ├── __init__.py
│   │   └── custom_middleware.py  # 自定义业务中间件
│   ├── models/              # 数据模型层（Pydantic）
│   │   ├── __init__.py
│   │   └── user.py          # 用户相关请求/响应模型
│   └── routers/             # 路由层
│       ├── __init__.py
│       ├── user.py          # 用户模块接口
│       └── upload.py        # 文件上传模块接口
└── requirements.txt         # 项目依赖清单

8.2 各模块完整代码

8.2.1 依赖清单 requirements.txt

中间件为框架内置能力，无新增依赖：

fastapi>=0.110.0
uvicorn>=0.27.0
python-multipart>=0.0.9
pydantic>=2.6.0

8.2.2 公共组件层

app/common/response.py

from pydantic import BaseModel
from typing import Generic, TypeVar, Optional

T = TypeVar("T")

class ResponseResult(BaseModel, Generic[T]):
    code: int = 200
    message: str = "success"
    data: Optional[T] = None

def success(data: T = None, message: str = "操作成功") -> ResponseResult[T]:
    return ResponseResult(code=200, message=message, data=data)

def fail(code: int = 400, message: str = "操作失败") -> ResponseResult:
    return ResponseResult(code=code, message=message, data=None)

app/common/exceptions.py

class BizException(Exception):
    """自定义业务异常类"""
    def __init__(self, code: int, message: str):
        self.code = code
        self.message = message
        super().__init__(message)

8.2.3 数据模型层

app/models/user.py

from pydantic import BaseModel, Field, field_validator

class UserCreate(BaseModel):
    username: str = Field(min_length=3, max_length=20, description="用户名")
    age: int = Field(ge=0, le=120, default=18, description="年龄")
    phone: str = Field(pattern=r"^1[3-9]\d{9}$", description="手机号")
    password: str = Field(min_length=6, max_length=32, description="密码")

    @field_validator("password")
    @classmethod
    def check_password(cls, value: str) -> str:
        if not any(c.isdigit() for c in value):
            raise ValueError("密码必须包含至少一位数字")
        return value

class UserResp(BaseModel):
    user_id: int
    username: str
    age: int
    phone: str

8.2.4 中间件层（新增）

app/middleware/custom_middleware.py
封装两个高频业务中间件，与入口文件解耦，符合工程化拆分规范：

import time
import logging
from fastapi import Request
from fastapi.responses import JSONResponse

logger = logging.getLogger(__name__)

# 鉴权白名单：无需登录即可访问的路径
AUTH_WHITE_LIST = [
    "/health",
    "/docs",
    "/openapi.json",
    "/redoc",
    "/user/list",
    "/user/create",
    "/upload/avatar"
]


async def access_log_middleware(request: Request, call_next):
    """
    全局访问日志中间件：统计接口耗时、记录请求全链路日志
    """
    start_time = time.perf_counter()
    # 将请求传递至下一层中间件/路由函数
    response = await call_next(request)
    # 计算接口执行耗时
    process_time = time.perf_counter() - start_time
    # 耗时写入响应头，前端可直接读取
    response.headers["X-Process-Time"] = f"{process_time:.4f}"
    # 控制台输出访问日志
    logger.info(
        f"{request.method} {request.url.path} "
        f"| 状态码: {response.status_code} "
        f"| 耗时: {process_time:.4f}s"
    )
    return response


async def auth_middleware(request: Request, call_next):
    """
    全局鉴权中间件：白名单路径直接放行，其余路径校验身份凭证
    """
    # 白名单路径跳过鉴权
    if request.url.path in AUTH_WHITE_LIST:
        return await call_next(request)

    # 从请求头提取 Token
    token = request.headers.get("Authorization")
    # 格式校验
    if not token or not token.startswith("Bearer "):
        return JSONResponse(
            status_code=401,
            content={"code": 401, "message": "缺少有效身份凭证", "data": None}
        )

    # 提取 Token 主体（实际项目替换为 JWT 解析、Redis 校验等业务逻辑）
    access_token = token.split(" ")[1]
    if access_token != "fastapi_demo_token":
        return JSONResponse(
            status_code=403,
            content={"code": 403, "message": "Token无效或已过期", "data": None}
        )

    # 鉴权通过，将用户信息注入 request.state，下游路由可直接读取
    request.state.user_id = 1001
    request.state.username = "demo_user"

    return await call_next(request)

8.2.5 路由层

app/routers/user.py（原有基础上新增鉴权测试接口）

from fastapi import APIRouter, Path, Query, Request
from app.models.user import UserCreate, UserResp
from app.common.response import ResponseResult, success
from app.common.exceptions import BizException

user_router = APIRouter(prefix="/user", tags=["用户管理"])


@user_router.get("/list", summary="获取用户列表", response_model=ResponseResult)
def get_user_list(
    page: int = Query(default=1, ge=1, description="页码"),
    page_size: int = Query(default=10, ge=1, le=100, description="每页数量")
):
    data = {
        "page": page,
        "page_size": page_size,
        "total": 100,
        "list": [{"user_id": i, "username": f"用户{i}"} for i in range(page_size)]
    }
    return success(data=data)


@user_router.get("/{user_id}", summary="获取用户信息", response_model=ResponseResult[UserResp])
def get_user(user_id: int = Path(..., gt=0, description="用户ID")):
    if user_id > 1000:
        raise BizException(code=1001, message="用户不存在")
    user_data = {
        "user_id": user_id,
        "username": f"用户{user_id}",
        "age": 20,
        "phone": "13800138000",
        "password": "123456"  # 响应模型自动过滤敏感字段
    }
    return success(data=user_data)


@user_router.post("/create", summary="创建用户", response_model=ResponseResult)
def create_user(user: UserCreate):
    # 模拟创建用户，返回自增ID
    new_user_id = 1001
    return success(data={"user_id": new_user_id}, message="用户创建成功")


@user_router.get("/me/info", summary="获取当前登录用户信息", response_model=ResponseResult)
def get_current_user(request: Request):
    """
    需鉴权接口：从 request.state 读取中间件注入的用户信息
    """
    user_info = {
        "user_id": request.state.user_id,
        "username": request.state.username
    }
    return success(data=user_info)

app/routers/upload.py

from fastapi import APIRouter, File, UploadFile
from app.common.response import ResponseResult, success

upload_router = APIRouter(prefix="/upload", tags=["文件上传"])

@upload_router.post("/avatar", summary="上传头像", response_model=ResponseResult)
async def upload_avatar(file: UploadFile = File(..., description="头像图片")):
    content = await file.read()
    data = {
        "filename": file.filename,
        "size": len(content),
        "content_type": file.content_type
    }
    return success(data=data, message="上传成功")

8.2.6 项目入口 main.py（整合中间件、异常、路由）

from fastapi import FastAPI, HTTPException, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from fastapi.middleware.cors import CORSMiddleware

from app.routers.user import user_router
from app.routers.upload import upload_router
from app.common.exceptions import BizException
from app.middleware.custom_middleware import access_log_middleware, auth_middleware

# 创建应用实例
app = FastAPI(
    title="FastAPI 全栈实战项目",
    description="整合路由、参数校验、异常处理、中间件、统一响应的工程化示例",
    version="1.0.0"
)

# ========== 1. 注册中间件（洋葱模型：先注册的在内层，后注册的在外层） ==========
# 1.1 CORS 跨域中间件（最外层，优先注册）
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:8080"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 1.2 全局日志 + 耗时统计中间件
app.middleware("http")(access_log_middleware)

# 1.3 全局鉴权中间件（最内层，最后注册）
app.middleware("http")(auth_middleware)

# ========== 2. 挂载业务路由 ==========
app.include_router(user_router)
app.include_router(upload_router)

# ========== 3. 全局异常处理器 ==========
@app.exception_handler(BizException)
async def biz_exception_handler(request: Request, exc: BizException):
    return JSONResponse(
        status_code=200,
        content={"code": exc.code, "message": exc.message, "data": None}
    )

@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    first_error = exc.errors()[0]
    error_msg = f"参数[{'.'.join(map(str, first_error['loc']))}]：{first_error['msg']}"
    return JSONResponse(
        status_code=400,
        content={"code": 400, "message": error_msg, "data": None}
    )

@app.exception_handler(HTTPException)
async def http_exception_handler(request: Request, exc: HTTPException):
    return JSONResponse(
        status_code=exc.status_code,
        content={"code": exc.status_code, "message": exc.detail, "data": None}
    )

@app.exception_handler(Exception)
async def global_exception_handler(request: Request, exc: Exception):
    return JSONResponse(
        status_code=500,
        content={"code": 500, "message": "服务器内部错误", "data": None}
    )

# ========== 4. 健康检查接口 ==========
@app.get("/health", summary="服务健康检查", tags=["系统接口"])
def health_check():
    return {"status": "ok", "service": "fastapi-demo"}

8.3 项目启动与测试

在项目根目录执行启动命令：

uvicorn app.main:app --reload

8.3.1 基础功能测试

# 1. 健康检查
curl http://127.0.0.1:8000/health

# 2. 获取用户列表（白名单接口，无需鉴权）
curl "http://127.0.0.1:8000/user/list?page=2&page_size=5"

# 3. 创建用户（白名单接口，无需鉴权）
curl -X POST http://127.0.0.1:8000/user/create \
  -H "Content-Type: application/json" \
  -d '{"username":"lisi","age":25,"phone":"13800138000","password":"abc123"}'

# 4. 文件上传
curl -X POST http://127.0.0.1:8000/upload/avatar \
  -F "file=@test.txt"

8.3.2 中间件能力测试

（1）验证耗时统计与日志

执行任意接口请求，查看响应头与控制台输出：

# 查看响应头中的耗时字段
curl -I http://127.0.0.1:8000/health

返回结果中会包含 X-Process-Time: 0.0005 类似字段，同时控制台自动输出访问日志。

（2）验证全局鉴权

# 不带Token访问需鉴权接口，返回401
curl http://127.0.0.1:8000/user/me/info

# 带错误Token访问，返回403
curl -H "Authorization: Bearer wrong_token" http://127.0.0.1:8000/user/me/info

# 带正确Token访问，正常返回用户信息
curl -H "Authorization: Bearer fastapi_demo_token" http://127.0.0.1:8000/user/me/info

（3）验证跨域配置

通过模拟预检请求验证跨域是否生效：

curl -X OPTIONS http://127.0.0.1:8000/health \
  -H "Origin: http://localhost:3000" \
  -H "Access-Control-Request-Method: GET"

返回头中包含 Access-Control-Allow-Origin 字段即代表配置生效。

---

还没结束，前面我们覆盖了路由、参数解析、校验、异常、统一响应、中间件这些工程化基础能力，接下来补充 FastAPI 最核心的进阶特性——依赖注入系统（Depends）。这是 FastAPI 区别于 Flask、Django 的核心优势之一，专门用来解决「接口间逻辑复用、业务解耦、权限粒度控制」的问题，也是企业级项目的标配写法。

---

九、依赖注入（Depends）：业务逻辑复用与解耦

9.1 核心概念

依赖注入（Dependency Injection）本质是「把接口需要的公共逻辑抽离出来，由框架自动注入到路由函数中」，不需要我们手动实例化、传参。

它解决的核心痛点：

• 避免多个接口重复编写相同的逻辑（比如分页参数解析、Token 鉴权、数据库连接）
• 实现接口粒度的权限控制，比全局中间件更灵活
• 业务逻辑分层清晰，便于测试和维护

FastAPI 的依赖注入通过 Depends() 实现，支持函数依赖、类依赖、嵌套依赖等多种形式。

9.2 基础语法：最简单的函数依赖

依赖本身就是一个普通的 Python 函数，接收的参数规则和路由函数完全一致（支持路径参数、查询参数、请求头等），路由函数通过 Depends(依赖函数) 声明即可自动注入结果。

from fastapi import FastAPI, Depends

app = FastAPI()

# 定义依赖函数：解析通用分页参数
def get_pagination(page: int = 1, page_size: int = 10):
    return {"page": page, "page_size": page_size}

# 在路由中声明依赖，框架自动执行依赖函数并把结果注入进来
@app.get("/items")
def get_items(pagination: dict = Depends(get_pagination)):
    return {
        "page": pagination["page"],
        "page_size": pagination["page_size"],
        "data": []
    }

效果等价于：你在路由函数里手动写了 page 和 page_size 两个查询参数，但通过依赖把这段逻辑抽离了出来，所有需要分页的列表接口都可以一键复用。

9.3 高频实战场景

9.3.1 场景一：公共参数复用（分页、筛选）

这是最基础也最常用的场景，把分页、排序、关键词筛选这类通用查询参数抽成依赖，所有列表类接口直接复用，消除重复代码。

from typing import Optional
from fastapi import Query, Depends

# 通用分页筛选依赖
def common_query_params(
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(10, ge=1, le=100, description="每页条数"),
    keyword: Optional[str] = Query(None, description="搜索关键词"),
    order_by: str = Query("id", description="排序字段")
):
    return {
        "page": page,
        "page_size": page_size,
        "keyword": keyword,
        "order_by": order_by
    }

# 用户列表接口直接复用
@app.get("/user/list")
def get_user_list(params: dict = Depends(common_query_params)):
    return {
        "code": 200,
        "message": "success",
        "data": params
    }

# 商品列表接口直接复用
@app.get("/goods/list")
def get_goods_list(params: dict = Depends(common_query_params)):
    return {
        "code": 200,
        "message": "success",
        "data": params
    }

测试命令：

curl "http://127.0.0.1:8000/user/list?page=2&page_size=20&keyword=test"

9.3.2 场景二：接口级鉴权依赖（官方推荐方案）

全局中间件会拦截所有请求，但实际业务中往往是「部分接口需要登录，部分不需要」。用依赖注入可以精准控制到单个接口，粒度更细，也是 FastAPI 官方推荐的鉴权实现方式。

from fastapi import Header, HTTPException, Depends

# 鉴权依赖：校验请求头中的 Token
def verify_token(authorization: str = Header(..., description="身份凭证")):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Token格式错误")
    
    token = authorization.split(" ")[1]
    # 实际项目替换为 JWT 解析、Redis 校验等业务逻辑
    if token != "fastapi_demo_token":
        raise HTTPException(status_code=403, detail="Token无效或已过期")
    
    # 校验通过，返回用户信息，路由函数可直接使用
    return {
        "user_id": 1001,
        "username": "demo_user"
    }

# 需要登录的接口：声明依赖即可自动完成鉴权
@app.get("/user/me", summary="获取当前用户信息（需登录）")
def get_my_info(current_user: dict = Depends(verify_token)):
    return {
        "code": 200,
        "message": "success",
        "data": current_user
    }

# 不需要登录的接口：不声明依赖，完全不受影响
@app.get("/public/list", summary="公开接口（无需登录）")
def get_public_list():
    return {"code": 200, "message": "success", "data": []}

测试命令：

# 不带 Token，直接返回 422/401 错误
curl http://127.0.0.1:8000/user/me

# 带正确 Token，正常返回用户信息
curl -H "Authorization: Bearer fastapi_demo_token" http://127.0.0.1:8000/user/me

和全局中间件的核心区别：

• 中间件：全局生效，适合日志、跨域这类全接口通用的基础逻辑
• 依赖注入：接口粒度生效，适合鉴权、权限校验这类差异化的业务逻辑
  实际项目中通常两者结合使用：中间件做全链路通用处理，依赖注入做业务级权限控制。

9.3.3 场景三：依赖嵌套

依赖之间可以互相嵌套，FastAPI 会自动解析完整的依赖链，非常适合分层的业务逻辑。

# 第一层依赖：从请求头提取原始 Token
def get_token(authorization: str = Header(...)):
    return authorization.split(" ")[1]

# 第二层依赖：依赖第一层的 Token，解析并返回用户信息
def get_current_user(token: str = Depends(get_token)):
    if token != "fastapi_demo_token":
        raise HTTPException(status_code=403, detail="Token无效")
    return {"user_id": 1001, "username": "demo"}

# 第三层：路由函数只依赖最终的用户信息
@app.get("/user/info")
def get_user_info(user: dict = Depends(get_current_user)):
    return {"code": 200, "data": user}

9.4 进阶：类作为依赖

除了函数，类也可以作为依赖，适合状态更复杂、需要封装方法的场景，比如数据库会话、配置读取、业务服务类等。

class UserService:
    """用户服务类：封装用户相关的业务逻辑"""
    def __init__(self, db_type: str = "mysql"):
        self.db_type = db_type
    
    def get_user_by_id(self, user_id: int):
        # 模拟数据库查询
        return {"user_id": user_id, "username": f"用户{user_id}", "db": self.db_type}

# 路由中注入类依赖，框架自动实例化
@app.get("/user/{user_id}")
def get_user(
    user_id: int,
    user_service: UserService = Depends(UserService)
):
    user = user_service.get_user_by_id(user_id)
    return {"code": 200, "data": user}

9.5 整合进实战项目

我们把依赖注入能力整合到前文的实战项目中，做两处核心优化：

1. 抽离通用分页依赖，替换用户列表接口的重复参数
2. 新增鉴权依赖，替换全局中间件的粗粒度鉴权，实现接口级精准控制

第一步：新增依赖文件 app/dependencies/common.py

# app/dependencies/common.py
from typing import Optional
from fastapi import Query, Header, HTTPException, Depends

def pagination_params(
    page: int = Query(1, ge=1, description="页码"),
    page_size: int = Query(10, ge=1, le=100, description="每页条数"),
    keyword: Optional[str] = Query(None, description="搜索关键词")
):
    """通用分页查询依赖"""
    return {
        "page": page,
        "page_size": page_size,
        "keyword": keyword
    }

def get_current_user(authorization: str = Header(..., description="身份凭证")):
    """鉴权依赖：校验Token并返回当前用户信息"""
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Token格式错误")
    
    token = authorization.split(" ")[1]
    if token != "fastapi_demo_token":
        raise HTTPException(status_code=403, detail="Token无效或已过期")
    
    return {
        "user_id": 1001,
        "username": "demo_user"
    }

第二步：修改用户路由 app/routers/user.py

from fastapi import APIRouter, Path, Depends
from app.models.user import UserCreate, UserResp
from app.common.response import ResponseResult, success
from app.common.exceptions import BizException
from app.dependencies.common import pagination_params, get_current_user

user_router = APIRouter(prefix="/user", tags=["用户管理"])

@user_router.get("/list", summary="获取用户列表", response_model=ResponseResult)
def get_user_list(params: dict = Depends(pagination_params)):
    # 直接使用注入的分页参数
    data = {
        "page": params["page"],
        "page_size": params["page_size"],
        "keyword": params["keyword"],
        "total": 100,
        "list": [{"user_id": i, "username": f"用户{i}"} for i in range(params["page_size"])]
    }
    return success(data=data)

@user_router.get("/{user_id}", summary="获取用户信息", response_model=ResponseResult[UserResp])
def get_user(user_id: int = Path(..., gt=0, description="用户ID")):
    if user_id > 1000:
        raise BizException(code=1001, message="用户不存在")
    user_data = {
        "user_id": user_id,
        "username": f"用户{user_id}",
        "age": 20,
        "phone": "13800138000",
        "password": "123456"
    }
    return success(data=user_data)

@user_router.post("/create", summary="创建用户", response_model=ResponseResult)
def create_user(user: UserCreate):
    new_user_id = 1001
    return success(data={"user_id": new_user_id}, message="用户创建成功")

@user_router.get("/me/info", summary="获取当前登录用户信息", response_model=ResponseResult)
def get_current_user_info(current_user: dict = Depends(get_current_user)):
    # 依赖自动完成鉴权，直接使用注入的用户信息
    return success(data=current_user)

第三步：调整 main.py 中间件配置

移除全局鉴权中间件，保留日志和跨域中间件，鉴权交由依赖注入精准控制：

# 只保留跨域和日志中间件，删除全局鉴权中间件
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000", "http://localhost:8080"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

app.middleware("http")(access_log_middleware)
# 删除 app.middleware("http")(auth_middleware) 这一行

改造后项目架构更合理：

• 跨域、访问日志：全局中间件统一处理
• 鉴权、分页参数：依赖注入按接口按需使用
• 业务逻辑分层清晰，后续扩展新模块时可直接复用公共依赖

---