02 - FastAPI 框架入门与原理

说到 Python 后端框架，FastAPI 这几年的势头确实猛。它诞生于 2018 年，由 Sebastián Ramírez 创建，目标很明确：要快、要简单、还要自带文档。今天这篇文章，我们就把它拆开来看一看，搞懂它到底是怎么回事，以及它凭什么比 Flask 快。

先列几个核心要点，我们一个一个来过：

• ASGI 和 WSGI 到底有什么本质区别
• FastAPI、Starlette、Pydantic 这三者是怎么分工的
• Depends 的依赖注入机制到底怎么跑
• 路由匹配和 APIRouter 如何组织
• lifespan 生命周期钩子（用来替代已经弃用的 on_event）
• FastAPI 为什么比 Flask 快

---

1. FastAPI 是什么

FastAPI 是一个现代 Python Web 框架，由 Sebastián Ramírez 于 2018 年创建。它有三大特点：

1. 快：基于 ASGI 异步，性能直追 Node.js 和 Go
2. 简单：类型注解驱动，让你少写很多代码
3. 自动文档：Swagger 和 ReDoc 都是自动生成的

1.1 三层依赖关系

 复制代码FastAPI
   │
   ├── Starlette（底层 ASGI 框架，负责路由、中间件、HTTP）
   │
   └── Pydantic（数据校验，把类型注解变成校验规则）

• Starlette：FastAPI 的 Web 内核，纯异步的 ASGI 框架
• Pydantic：数据校验库，FastAPI 用它来校验请求体、序列化响应
• FastAPI：在两者之上加了一层语法糖，让开发体验更爽

1.2 一个最简单的 FastAPI 应用

 复制代码from fastapi import FastAPIapp = FastAPI()@app.get("/")
def hello():
    return {"msg": "hello"}

启动命令也很简单：

 复制代码uvicorn main:app --reload

这里的 main:app 表示 main.py 文件里的 app 变量。

---

2. ASGI vs WSGI：为什么 FastAPI 快

2.1 WSGI（同步）

老一代的 Python Web 框架（比如 Flask、Django）走的是 WSGI 协议：

 复制代码请求 → WSGI Server (gunicorn) → Flask → 处理 → 响应

这里面有个问题：一个请求占用一个 worker，等数据库查询的时候，worker 就闲着，白白浪费资源。

2.2 ASGI（异步）

 复制代码请求 → ASGI Server (uvicorn) → FastAPI (async) → 处理（可让出 CPU）→ 响应

优势就在于异步 I/O：一个 worker 可以同时处理多个请求。等数据库的时候可以主动让出 CPU，去处理其他请求。

2.3 同步 vs 异步代码对比

 复制代码# Flask（同步）
@app.route("/users")
def get_users():
    users = db.query_all()  # 阻塞等数据库，worker 空闲
    return jsonify(users)# FastAPI（异步）
@app.get("/users")
async def get_users():
    users = await db.query_all()  # 等数据库时让出 CPU
    return users

2.4 本项目为什么用同步

你看项目里的代码：

 复制代码@router.get("")
def list_users(svc: UserService = Depends(get_user_service)):  # def 不是 async def
    return svc.list_users()

这里用了 def 而不是 async def，原因很简单：SQLAlchemy 2.0 的同步 API 是阻塞的。FastAPI 会自动把同步路由放到线程池里去执行，不会阻塞事件循环。

写法	适合场景
async def	I/O 异步库（asyncpg、httpx 异步、Redis 异步）
def	同步库（SQLAlchemy 同步、requests）

本项目用了 SQLAlchemy 同步 API，所以路由统一用 def。

---

3. 依赖注入：Depends 的原理

依赖注入可以说是 FastAPI 最强大的特性。先看个用法：

 复制代码def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()@router.get("/users")
def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

Depends(get_db) 的意思是告诉 FastAPI：“在调用 list_users 之前，先调用 get_db，然后把结果作为 db 参数传进去”。

3.1 yield 依赖的执行流程

 复制代码def get_db():
    db = SessionLocal()    # 1. 创建 Session
    try:
        yield db           # 2. 把 db 注入路由
    finally:
        db.close()         # 3. 路由执行完，关闭 Session

执行顺序大致是这样的：

 复制代码请求进来
  ↓
FastAPI 调 get_db()
  ↓
get_db 执行到 yield，把 db 给路由
  ↓
路由执行（用 db 查数据库）
  ↓
路由返回
  ↓
FastAPI 继续 get_db 的 finally，关闭 db
  ↓
返回响应给客户端

关键点在于：不管路由是成功返回还是中途抛异常，finally 都会被执行，这样就保证了数据库连接一定会被释放。

3.2 嵌套依赖

依赖还可以嵌套：

 复制代码def get_db():
    db = SessionLocal()
    yield dbdef get_user_service(db: Session = Depends(get_db)):
    return UserService(UserRepo(db))@router.get("/users")
def list_users(svc: UserService = Depends(get_user_service)):
    return svc.list_users()

FastAPI 会自动按依赖顺序解析：

 复制代码get_db → 拿到 db
   ↓
get_user_service(db) → 拿到 svc
   ↓
list_users(svc)

3.3 依赖缓存（同请求内）

有意思的是，同一个请求里，Depends(get_db) 只会调用一次：

 复制代码@router.get("/")
def index(
    db1: Session = Depends(get_db),
    db2: Session = Depends(get_db),  # 同一个 db 实例
):
    assert db1 is db2  # True

这么做的好处是：一个请求内共享同一个 db session，避免了重复创建的开销。

3.4 依赖的作用

场景	依赖做什么
数据库 Session	每请求独立 session
当前用户	从 JWT 解析当前用户
权限校验	检查是否有权限
分页参数	解析 skip/limit
限流	检查调用频率

3.5 Depends vs 普通函数调用

你可能会问，为什么不直接调用函数？

 复制代码#  直接调用
@router.get("/users")
def list_users():
    db = SessionLocal()  # 没法自动关闭
    try:
        return db.query(User).all()
    finally:
        db.close()

这样写有几个问题：

1. 每个路由都要重复写这些代码
2. 异常处理时 session 关闭逻辑比较复杂
3. 测试时很难 mock

而用 Depends 的方式：

1. 一个依赖可以在所有路由中复用
2. 异常时自动处理（finally 保证）
3. 测试时替换依赖很方便：

 复制代码app.dependency_overrides[get_db] = get_test_db

---

4. 路由匹配与 APIRouter

4.1 路由装饰器

 复制代码@app.get("/")           # GET 请求
@app.post("/users")     # POST 请求
@app.put("/users/{id}") # PUT 请求
@app.delete("/users/{id}")  # DELETE 请求

每个装饰器对应一个 HTTP 方法，很直观。

4.2 路径参数

 复制代码@app.get("/users/{user_id}")
def get_user(user_id: int):  # 类型注解自动校验
    return {"id": user_id}

这里 user_id: int 的作用是让 FastAPI 自动完成三件事：

1. 从 URL 中解析出 user_id
2. 将它转为 int 类型（转不了就返回 422 错误）
3. 传给路由函数

4.3 查询参数

 复制代码@app.get("/users")
def list_users(skip: int = 0, limit: int = 20):
    return {"skip": skip, "limit": limit}

URL 里没有在路径中间出现的参数（比如 skip、limit），会自动当作查询参数处理：/users?skip=0&limit=20。

4.4 用 Query 加约束

 复制代码from fastapi import Query@router.get("")
def list_users(
    skip: int = Query(0, ge=0, description="跳过条数"),
    limit: int = Query(20, ge=1, le=100, description="每页数量"),
):
    ...

• ge=0：参数值必须 ≥ 0
• le=100：参数值必须 ≤ 100
• description：会在 Swagger 文档里显示

4.5 APIRouter：模块化路由

在大型项目里，不会把所有路由都堆在一个文件里。用 APIRouter 可以很好地拆分：

 复制代码# app/api/users.py
from fastapi import APIRouterrouter = APIRouter(prefix="/users", tags=["用户管理"])@router.get("")
def list_users(): ...@router.get("/{user_id}")
def get_user(user_id: int): ...

 复制代码# main.py
from fastapi import FastAPI
from app.api import users_routerapp = FastAPI()
app.include_router(users_router)
# 所有 /users 路径都由 users_router 处理

这样做的好处很明显：

• 不同资源拆到不同文件（users / orders / products）
• prefix 自动加前缀，避免重复写路径
• tags 在 Swagger 里按组显示，清晰明了

4.6 路由匹配顺序（重要坑）

 复制代码@router.get("/users/me")      # ① 先注册
@router.get("/users/{user_id}")  # ② 后注册

FastAPI 是按注册顺序来匹配路由的。如果顺序反过来：

 复制代码@router.get("/users/{user_id}")  # 先注册
@router.get("/users/me")          # 后注册

这时访问 /users/me 会匹配到第一个路由，把 me 当作 user_id，然后尝试转 int 类型失败，返回 422 错误。

所以核心规则是：静态路径写在前，动态路径写在后。

---

5. 生命周期：lifespan vs on_event

5.1 弃用的 on_event

以前的写法是这样的：

 复制代码@app.on_event("startup")
def on_startup():
    Base.metadata.create_all(bind=engine)@app.on_event("shutdown")
def on_shutdown():
    engine.dispose()

但从 FastAPI 0.93 开始，这种写法已经被弃用了，未来会被移除。

5.2 新写法：lifespan

 复制代码from contextlib import asynccontextmanager@asynccontextmanager
async def lifespan(app: FastAPI):
    # ── startup 阶段 ──
    Base.metadata.create_all(bind=engine)
    yield  # 应用运行期间挂起
    # ── shutdown 阶段 ──
    engine.dispose()app = FastAPI(lifespan=lifespan)

5.3 asynccontextmanager 原理

asynccontextmanager 的作用是把一个生成器函数转成异步上下文管理器。

 复制代码@asynccontextmanager
async def lifespan(app):
    print("startup")  # yield 之前 = __aenter__
    yield
    print("shutdown")  # yield 之后 = __aexit__

执行流程是这样的：

 复制代码应用启动
  ↓
调用 lifespan(app)
  ↓
执行到 yield 之前的代码（startup）
  ↓
yield 让出控制权，应用开始接收请求
  ↓
（应用运行，处理无数请求）
  ↓
应用收到关闭信号（Ctrl+C / SIGTERM）
  ↓
继续执行 yield 之后的代码（shutdown）
  ↓
应用退出

5.4 为什么用 lifespan 而不是 on_event

维度	on_event	lifespan
状态	已弃用	推荐
共享状态	麻烦（用全局变量）	简单（yield 之前的变量可传递）
资源管理	分散（startup 和 shutdown 是两个函数）	集中（一个函数里）
异步支持	一般	原生

5.5 本项目的 lifespan

 复制代码@asynccontextmanager
async def lifespan(app: FastAPI):
    Base.metadata.create_all(bind=engine)
    logger.info("数据库表已就绪")
    yield
    engine.dispose()
    logger.info("应用已关闭，连接池释放")app = FastAPI(lifespan=lifespan)

---

6. 请求处理流程

一个完整的请求处理流程大致是这样：

 复制代码1. 客户端发起 HTTP 请求
   ↓
2. ASGI Server (uvicorn) 接收
   ↓
3. Starlette 中间件链（CORS、认证等）
   ↓
4. FastAPI 路由匹配
   ↓
5. 依赖注入解析（Depends 链）
   ↓
6. Pydantic 校验请求（body / query / path 参数）
   ↓
7. 调用路由函数
   ↓
8. 业务逻辑执行（Service → Repository → DB）
   ↓
9. Pydantic 序列化响应（response_model）
   ↓
10. 中间件链（响应方向）
   ↓
11. ASGI Server 返回响应

6.1 Pydantic 校验失败 → 422

如果请求体不符合 Schema 定义，FastAPI 会自动返回 422 错误：

 复制代码POST /users
{"username": "ab"}  ← 长度不够 3响应 422:
{
  "detail": [
    {
      "type": "string_too_short",
      "loc": ["body", "username"],
      "msg": "String should ha ve at least 3 characters"
    }
  ]
}

6.2 response_model 过滤字段

 复制代码@router.post("", response_model=UserOut)
def create_user(payload: UserCreate, ...):
    return svc.create_user(payload)  # 返回 UserOut

即使 service 返回的是 UserInDB（里面包含了 hashed_password），FastAPI 会按照 UserOut 的字段定义来过滤响应，这就相当于双重保险，防止敏感信息泄露。

---

7. 异常处理

7.1 HTTPException

 复制代码from fastapi import HTTPException, status@router.get("/{user_id}")
def get_user(user_id: int):
    user = db.get(user_id)
    if user is None:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="用户不存在"
        )
    return user

FastAPI 会自动把 HTTPException 转成对应的 HTTP 响应。

7.2 全局异常处理器（本项目用法）

 复制代码@app.exception_handler(UserNotFoundError)
async def handle_user_not_found(_, exc):
    return JSONResponse(status_code=404, content={"detail": str(exc)})@app.exception_handler(Exception)
async def handle_unexpected_error(_, exc):
    logger.exception(...)
    return JSONResponse(status_code=500, content={"detail": "服务器内部错误"})

这样做的好处是：路由函数里不需要再写 try/except，业务异常直接抛就行。

7.3 status 常量

 复制代码from fastapi import statusstatus.HTTP_200_OK          # 200
status.HTTP_201_CREATED     # 201
status.HTTP_204_NO_CONTENT  # 204
status.HTTP_400_BAD_REQUEST # 400
status.HTTP_404_NOT_FOUND   # 404
status.HTTP_409_CONFLICT    # 409
status.HTTP_422_UNPROCESSABLE_ENTITY  # 422
status.HTTP_500_INTERNAL_SERVER_ERROR # 500

用常量代替硬编码的数字，可读性会好很多。

---

8. FastAPI vs Flask 全面对比

维度	Flask	FastAPI
协议	WSGI（同步）	ASGI（异步）
性能	较慢	快（接近 Node.js）
类型注解	可选	核心
数据校验	需插件	内置（Pydantic）
自动文档	需 flask-restx	内置 Swagger + ReDoc
异步支持	需 async-flask	原生
学习曲线	简单	简单（类型注解驱动）
生态	极其丰富	快速增长

8.1 同接口代码对比

Flask：

 复制代码from flask import Flask, request, jsonifyapp = Flask(__name__)@app.route("/users", methods=["POST"])
def create_user():
    data = request.get_json()
    if not data.get("username") or len(data["username"]) < 3:
        return jsonify({"error": "用户名至少 3 位"}), 400
    if not data.get("password") or len(data["password"]) < 6:
        return jsonify({"error": "密码至少 6 位"}), 400
    # ... 手动校验
    user = create_in_db(data)
    return jsonify({"id": user.id, "username": user.username}), 201

FastAPI：

 复制代码from fastapi import FastAPI
from pydantic import BaseModel, Fieldapp = FastAPI()class UserCreate(BaseModel):
    username: str = Field(..., min_length=3)
    password: str = Field(..., min_length=6)@app.post("/users", status_code=201)
def create_user(payload: UserCreate):
    user = create_in_db(payload)
    return {"id": user.id, "username": user.username}

对比下来很明显：FastAPI 靠类型注解和 Pydantic 来自动校验，代码量直接少了一半。

---

9. 自动文档

启动服务后，访问：

• - Swagger UI（可以在线测试）
• - ReDoc（只读文档）

9.1 文档元信息

 复制代码@router.post(
    "",
    response_model=UserOut,
    status_code=status.HTTP_201_CREATED,
    summary="创建用户",
    description="创建新用户，用户名唯一。冲突返回 409。",
    tags=["用户管理"],
)
def create_user(...): ...

这些元信息都会在 Swagger 文档里清晰展示出来。

9.2 为什么用文档

• 前端联调：直接给前端 Swagger URL，可以自动生成 API client
• 测试：Swagger UI 能直接发请求，非常方便
• 新人入门：看文档比翻代码快得多

---

10. 常见坑

10.1 同步路由阻塞事件循环

 复制代码#  同步阻塞操作放在 async def 里
@app.get("/slow")
async def slow():
    time.sleep(5)  # 阻塞整个事件循环 5 秒
    return {"ok": True}

修复方法：用 def 让 FastAPI 放到线程池执行，或者用 asyncio.sleep：

 复制代码#  方案 1：用 def
@app.get("/slow")
def slow():
    time.sleep(5)
    return {"ok": True}#  方案 2：用异步 sleep
@app.get("/slow")
async def slow():
    await asyncio.sleep(5)
    return {"ok": True}

10.2 忘记 await

 复制代码#  忘记 await
@app.get("/")
async def index():
    result = some_async_func()  # 返回 coroutine，没执行
    return result#  正确写法
@app.get("/")
async def index():
    result = await some_async_func()
    return result

10.3 路由顺序

 复制代码#  动态路由在前
@router.get("/{user_id}")
@router.get("/me")  # 永远匹配不到#  静态路由在前
@router.get("/me")
@router.get("/{user_id}")

10.4 response_model 漏字段

 复制代码class UserOut(BaseModel):
    id: int
    username: str@router.get("/{user_id}", response_model=UserOut)
def get_user(user_id: int):
    return {"id": 1, "username": "alice", "password": "xxx"}  # 多余字段# 实际响应：{"id": 1, "username": "alice"}  ← password 被过滤

这其实是件好事，但前提是得记得定义好 response_model。

---

11. 自测题

Q1：下面代码有什么问题？

 复制代码@app.get("/users/{user_id}")
async def get_user(user_id):
    return {"id": user_id}

查看答案

user_id 没有类型注解，FastAPI 不知道怎么解析。应该写 user_id: int。

Q2：Depends(get_db) 用了 yield，什么时候执行 finally？

查看答案

路由函数执行完毕（无论成功还是异常）后，FastAPI 会继续执行 yield 之后的代码，包括 finally 块。

Q3：下面两个路由，访问 /users/me 会匹配哪个？

 复制代码@router.get("/users/{user_id}")
def get_user(user_id: int): ...@router.get("/users/me")
def get_me(): ...

查看答案

匹配第一个 /{user_id}，把 me 当 user_id，转 int 失败返回 422。应该把 /users/me 放在前面。

---

12. 小结

概念	关键点
ASGI	异步协议，I/O 等待时让出 CPU
FastAPI 架构	Starlette（Web）+ Pydantic（校验）
Depends	依赖注入，自动调用、自动关闭、可嵌套
APIRouter	模块化路由，prefix 加前缀
lifespan	替代弃用的 on_event，集中管理生命周期
response_model	自动序列化 + 字段过滤
全局异常处理	@app.exception_handler 注册

13. 下篇预告

下一篇会深入讲解 Pydantic v2 数据校验：BaseModel 原理、Field 约束、model_validate vs model_dump、ConfigDict、自定义校验器、泛型模型。

---

延伸阅读：

• FastAPI 官方文档
• Starlette 官方文档
• ASGI 规范
• PEP 3333 - WSGI