引言
FastAPI 这几年很受 Python 后端开发者欢迎，不只是因为它“快”，更因为它把现代 Web API 开发里几个痛点处理得很漂亮：
异步支持：天然支持
async def，适合数据库、缓存、HTTP 调用这类 IO 密集型服务。自动文档生成：基于类型标注和 Pydantic 模型生成 OpenAPI 文档，接口可读、可调试。
依赖注入：用
Depends管理数据库会话、鉴权、分页参数等公共逻辑。类型友好：参数校验、编辑器提示、文档展示基本都围绕类型声明展开。
我们可以看到一个典型学习型示例：它把 FastAPI 路由、SQLAlchemy 2.x 异步 ORM、MySQL 连接池、事务提交与回滚都放在了一起。对于刚开始做后端项目的同学来说，这正好是一条很实用的主线。
项目代码在做什么
db/orm.py 的核心结构可以概括为下面这张表：
我们先看整体请求链路：
flowchart LR
A["客户端请求 /db/book/add"] --> B["FastAPI 路由匹配"]
B --> C["Pydantic 校验 BookCreate"]
C --> D["Depends(get_db) 创建 AsyncSession"]
D --> E["执行业务逻辑和 SQLAlchemy ORM"]
E --> F{"是否异常?"}
F -->|否| G["commit 提交事务"]
F -->|是| H["rollback 回滚事务"]
G --> I["返回 JSON 响应"]
H --> J["返回错误响应"]我的实战感悟：FastAPI 项目写久了你会发现，真正决定代码是否好维护的，不是某个接口写得多炫，而是“请求参数、业务逻辑、数据库会话、响应模型”这些边界是否清楚。
FastAPI 核心特性
1. 异步支持：为什么用 async def
在传统同步 Web 框架里，一个请求访问数据库时，线程会等待数据库返回结果。FastAPI 基于 ASGI，可以用 async def 把等待 IO 的时间让出来，让事件循环继续处理其他请求。
这并不意味着单条 SQL 会变快。它真正提升的是：
大量并发请求下的连接利用率。
等待数据库、Redis、第三方 API 时的整体吞吐。
Web 服务在 IO 密集型场景下的响应能力。
这些接口都是异步的：
@router.get("/book/detail")
async def read_book_detail(...):
...只要内部调用的是异步数据库驱动，例如 mysql+aiomysql，就可以通过 await db.execute(...) 非阻塞地等待结果。
2. 自动文档生成：类型标注就是接口说明
FastAPI 会根据函数签名、Query、Body、Pydantic 模型自动生成 OpenAPI 文档。启动项目后，我们通常可以访问：
http://127.0.0.1:8000/docshttp://127.0.0.1:8000/redoc
比如下面这个参数：
book_id: Annotated[int, Query(description="图书 ID")]FastAPI 会知道它是一个查询参数，类型是整数，并把描述展示在接口文档里。
3. 依赖注入：把公共逻辑抽出来
数据库会话、当前登录用户、分页参数、权限校验，都可以用 Depends 注入。它的好处是：
路由函数只关注业务逻辑。
公共资源可以统一创建和释放。
测试时更容易替换依赖。
get_db() 就是一个很典型的依赖：
async def get_db():
async with async_session() as session:
try:
# 使用yield返回会话对象，等待调用方使用。
# 使用return返回会话对象，会立即关闭会话对象(需要每一次在路由函数中commit、rollback、close ，而且容易忘)
yield session
await session.commit()
except Exception:
await session.rollback()
raise这个依赖在请求正常结束后提交事务，在出现异常时回滚事务。路由函数只需要这样使用：
async def read_book_count(db: AsyncSession = Depends(get_db)):
...请求体验证：用 Pydantic 把错误挡在入口
定义了 BookCreate：
class BookCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=10, description="书名")
price: float = Field(..., description="价格")
author: str = Field(..., description="作者")
description: str = Field(..., description="描述")这段代码的价值不是“少写几个 if”，而是把接口契约明确写出来：调用方必须传什么字段、字段类型是什么、长度限制是什么。
第二个可运行示例：
# validation_demo.py
from typing import Annotated
from fastapi import Body, FastAPI
from pydantic import BaseModel, Field
app = FastAPI(title="Book Validation Demo")
class BookCreate(BaseModel):
name: str = Field(..., min_length=1, max_length=10, description="书名")
price: float = Field(..., gt=0, description="价格必须大于 0")
author: str = Field(..., min_length=1, description="作者")
description: str = Field(default="", max_length=255, description="描述")
@app.post("/books")
async def create_book(
book: Annotated[BookCreate, Body(description="图书信息")],
) -> dict[str, object]:
# 请求体已经被 FastAPI + Pydantic 校验过
return {"message": "created", "book": book.model_dump()}运行：
uvicorn validation_demo:app --reload请求示例：
{
"name": "Python 入门",
"price": 59.9,
"author": "Alice",
"description": "适合初学者"
}如果 price 传 -1，FastAPI 会直接返回 422 Unprocessable Entity，业务函数甚至不会被执行。
依赖注入：请求级数据库会话
在实际项目里，数据库连接不应该在每个路由里手动创建。更推荐的方式是：
应用启动时创建
engine和sessionmaker。每个请求通过依赖创建一个
AsyncSession。请求成功提交，失败回滚。
应用关闭时释放连接池。
第三个可运行示例基于 SQLite，方便你本地直接体验异步依赖注入，不需要先安装 MySQL。
# dependency_demo.py
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from typing import Annotated
from fastapi import Depends, FastAPI
from sqlalchemy import Integer, String, select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
engine = create_async_engine("sqlite+aiosqlite:///./demo.db", echo=True)
SessionLocal = async_sessionmaker(engine, expire_on_commit=False)
class Base(DeclarativeBase):
pass
class Book(Base):
__tablename__ = "books"
id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(50), unique=True)
async def get_session() -> AsyncIterator[AsyncSession]:
async with SessionLocal() as session:
try:
yield session
await session.commit()
except Exception:
await session.rollback()
raise
@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncIterator[None]:
# 启动时建表，关闭时释放连接池
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield
await engine.dispose()
app = FastAPI(title="Dependency Injection Demo", lifespan=lifespan)
@app.post("/books")
async def add_book(
name: str,
session: Annotated[AsyncSession, Depends(get_session)],
) -> dict[str, object]:
book = Book(name=name)
session.add(book)
await session.flush()
return {"id": book.id, "name": book.name}
@app.get("/books")
async def list_books(
session: Annotated[AsyncSession, Depends(get_session)],
) -> list[dict[str, object]]:
result = await session.execute(select(Book).order_by(Book.id.desc()))
books = result.scalars().all()
return [{"id": book.id, "name": book.name} for book in books]需要依赖：
pip install fastapi "uvicorn[standard]" "sqlalchemy[asyncio]" aiosqlite运行：
uvicorn dependency_demo:app --reload异步 ORM 如何工作
1. 创建异步 Engine
项目里使用的是：
ASYNC_DATABASE_URL = (
"mysql+aiomysql://root:******@localhost:3306/fastapi_test?charset=utf8"
)
async_engine = create_async_engine(
ASYNC_DATABASE_URL,
echo=True,
pool_size=10,
max_overflow=20,
)这里有三个关键点：
mysql+aiomysql表示使用 MySQL 的异步驱动。//后跟账号密码还有地址已经数据库名称(注意数据库在连接的时候需要提前创建)
echo=True会打印 SQL，适合开发调试，不建议生产长期打开。pool_size和max_overflow控制连接池容量，不能盲目调大。
最佳实践：数据库账号、密码、主机、库名建议放到环境变量或
.env中，不要硬编码在源码里。尤其是教学代码进入真实项目时，这一步要尽早做。
2. 定义 ORM 模型
Book 继承了一个包含时间字段的基类：
class DATA_TIME(DeclarativeBase):
create_time: Mapped[datetime] = mapped_column(
DateTime,
insert_default=func.now(),
default=func.now(),
comment="创建时间",
)
update_time: Mapped[datetime] = mapped_column(
DateTime,
insert_default=func.now(),
default=func.now(),
onupdate=func.now(),
comment="修改时间",
)Mapped 和 mapped_column 是 SQLAlchemy 2.x 推荐的 ORM 声明方式。它比老式写法更贴近 Python 类型系统，编辑器提示也更舒服。
图书表可以理解为：
class Book(DATA_TIME):
__tablename__ = "book"
id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
name: Mapped[str] = mapped_column(String(255), unique=True)
author: Mapped[str] = mapped_column(String(255))
price: Mapped[float] = mapped_column(Float)
description: Mapped[str] = mapped_column(String(255))这里 name 设置了 unique=True，意味着数据库层面不允许书名重复。代码中 add_book() 也提前查了一次重名，用来给用户返回更友好的错误。
3. 创建表结构
async def create_db_and_tables():
async with async_engine.begin() as conn:
await conn.run_sync(DATA_TIME.metadata.create_all)run_sync() 是异步 SQLAlchemy 里常见的桥接方法：有些元数据操作本质上还是同步 API，需要通过异步连接安全地运行。
生产项目里，我更建议用 Alembic 管理表结构迁移。metadata.create_all() 很适合学习和原型开发，但不适合长期维护复杂数据库结构。
4. 管理事务边界
get_db() 的写法体现了一个很重要的思想：一个请求，一个会话，一个事务边界。
sequenceDiagram
participant Client as Client
participant API as FastAPI Endpoint
participant Dep as get_db()
participant DB as MySQL
Client->>API: HTTP Request
API->>Dep: Depends(get_db)
Dep->>DB: Open AsyncSession
API->>DB: execute / add / delete
alt success
Dep->>DB: commit()
else exception
Dep->>DB: rollback()
end
DB-->>Client: JSON Response这样设计的好处是：路由函数不用反复写 commit() / rollback()，并且异常场景不会把半完成的数据留在数据库里。
CRUD 实战：图书接口怎么写
查询单条：get() 和 execute() 的区别
db.get(Book, 1) 适合按主键查询：
book = await db.get(Book, 1)如果你要按条件查询，就使用 select()：
res = await db.execute(select(Book).where(Book.id == book_id))
book = res.scalar_one_or_none()两者的使用场景如下：
条件查询：组合表达式要注意括号
项目里有类似这样的条件：
select(Book).where(
Book.name.like(f"%{name}%") & (Book.price > 100)
| (Book.author == "张三") & ~(Book.description == "无")
)为了可读性，我建议写成：
select(Book).where(
(
Book.name.like(f"%{name}%")
& (Book.price > 100)
)
| (
(Book.author == "张三")
& ~(Book.description == "无")
)
)&、|、~ 分别表示 SQL 条件里的 AND、OR、NOT。因为 Python 运算符优先级容易让人看错，复杂条件最好主动加括号。
分页查询：先满足可用，再考虑性能
项目中分页接口使用 offset + limit：
res = await db.execute(
select(Book)
.offset((current - 1) * size)
.limit(size)
)
return res.scalars().all()这是最容易理解的分页方式，适合数据量不大的后台列表。等数据量上来后，再考虑基于游标的分页，例如 id < last_id。
新增图书：为什么要 flush() 和 refresh()
db_book = Book(**book.model_dump())
db.add(db_book)
await db.flush()
await db.refresh(db_book)
return db_book这里的重点是：
add()：把对象加入 Session。flush()：把 SQL 发到数据库，但还没有最终提交事务。refresh()：从数据库重新加载对象，拿到自增 ID、默认时间等字段。commit()：在get_db()依赖中统一执行。
也就是说，flush() 不是提交，它只是把当前变更同步到数据库事务里。
更新和删除：先查存在性
更新：
db_book = await db.get(Book, book_id)
if db_book is None:
raise HTTPException(status_code=404, detail="图书不存在")
db_book.name = book.name
db_book.price = book.price
db_book.author = book.author
db_book.description = book.description
await db.flush()
await db.refresh(db_book)
return db_book删除：
db_book = await db.get(Book, book_id)
if db_book is None:
raise HTTPException(status_code=404, detail="图书不存在")
await db.delete(db_book)
await db.flush()
return {"msg": "删除成功"}我的习惯是：更新、删除都先查对象是否存在。这样接口返回会更清楚，也方便后续加权限判断、审计日志等业务逻辑。
自动文档：让接口天然可调试
FastAPI 的自动文档来自三个地方：
比如分页接口：
@router.get("/book/page")
async def read_book_page(
current: int = Query(1, ge=1, description="当前页码"),
size: int = Query(10, ge=1, description="每页数量"),
db: AsyncSession = Depends(get_db),
):
...在 /docs 里，current 和 size 会自动显示默认值、最小值和描述。对于前后端协作来说，这比单独维护一份接口文档省心很多。
最佳实践清单
1. 路由层不要变成“大杂烩”
当前 db/orm.py 为了学习方便，把模型、数据库连接、依赖、路由都放在了一个文件里。真实项目建议拆分：
app/
db/
engine.py # engine 和 sessionmaker
models.py # ORM 模型
dependencies.py # get_db 依赖
schemas/
book.py # Pydantic 请求/响应模型
routers/
book.py # API 路由拆分不是为了“显得高级”，而是为了让每个文件职责更单一。等项目变大时，你会感谢自己早一点做了这件事。
2. 不要在源码里硬编码数据库密码
推荐使用环境变量：
import os
DATABASE_URL = os.environ["DATABASE_URL"]启动时传入：
DATABASE_URL="mysql+aiomysql://user:password@127.0.0.1:3306/app?charset=utf8mb4" uvicorn main:app --reload另外，MySQL 字符集建议优先使用 utf8mb4，比 utf8 更完整，能正确支持 emoji 和更多 Unicode 字符。
3. 给接口加 response_model
直接返回 ORM 对象在学习阶段很方便，但真实项目里我更推荐加响应模型：
from pydantic import BaseModel
class BookRead(BaseModel):
id: int
name: str
price: float
author: str
description: str
@router.get("/book/detail", response_model=BookRead | None)
async def read_book_detail(...):
...这样可以避免把不该暴露的字段返回给客户端，也能让 OpenAPI 文档更准确。
4. 生产环境使用 Alembic 管理迁移
create_all() 适合入门学习，但生产环境需要可追踪、可回滚、可审核的迁移脚本。推荐引入 Alembic：
alembic init migrations
alembic revision --autogenerate -m "create book table"
alembic upgrade head5. 事务要短，外部 IO 不要塞进事务
下面这种模式要尽量避免：
开启事务 -> 查数据库 -> 调第三方接口 -> 写数据库 -> 提交事务第三方接口慢的时候，数据库连接和锁都会被长期占用。更好的做法是：事务里只做必要的数据库操作，外部 IO 放到事务外，复杂场景再考虑消息队列或 outbox pattern。
常见问题
flush() 和 commit() 有什么区别
flush() 是把变更发送到数据库事务里，常用于拿自增 ID 或触发约束检查；commit() 是真正提交事务。一个请求里可以多次 flush()，但通常只在事务边界提交一次。
为什么 expire_on_commit=False
异步 ORM 下，提交后如果对象属性过期，再访问属性可能触发隐式 IO。隐式 IO 在异步环境里容易引发 MissingGreenlet 一类问题。设置 expire_on_commit=False 能减少这种意外。
为什么我建议显式写 Query、Body
因为它们不仅影响运行时校验，也会影响自动文档。对团队项目来说，文档准确本身就是生产力。
小结
通过 db/orm.py，我们其实已经把 FastAPI 后端项目里最重要的一组能力串起来了：
用
APIRouter组织接口模块。用
async def和异步数据库驱动支撑高并发 IO。用 Pydantic 把请求体校验前置。
用
Depends(get_db)管理请求级数据库会话。用 SQLAlchemy 2.x ORM 完成增删改查。
用
HTTPException表达业务错误。
如果你刚开始学习 FastAPI，我建议先把这套图书 CRUD 跑通，再逐步把配置、模型、Schema、路由拆分出去。先跑通，再变优雅，这是我自己做后端项目时一直很相信的节奏。
参考资料
FastAPI 官方文档：https://fastapi.tiangolo.com/
FastAPI Dependencies：https://fastapi.tiangolo.com/tutorial/dependencies/
FastAPI Body - Fields：https://fastapi.tiangolo.com/tutorial/body-fields/
SQLAlchemy asyncio 文档：https://docs.sqlalchemy.org/en/20/orm/extensions/asyncio.html
SQLAlchemy ORM Querying Guide：https://docs.sqlalchemy.org/en/20/orm/queryguide/
SQLAlchemy Session Basics：https://docs.sqlalchemy.org/en/20/orm/session_basics.html