【python】【FastAPI 进阶】《FastAPI 异步内核剖析：中间件链与 SQLAlchemy 异步驱动的全链路非阻塞实践》

m0_74821874

414人浏览 · 2026-08-13 18:48:14

m0_74821874 · 2026-08-13 18:48:14 发布

FastAPI 中间件、依赖注入与 SQLAlchemy 异步 ORM

第一章整体架构与核心概念

FastAPI 的请求处理可以抽象为一条完整链路：客户端发出 HTTP 请求，请求经过中间件后进入路由；FastAPI 解析路由参数并解析依赖关系；数据库依赖创建异步会话；路由通过 SQLAlchemy ORM 执行查询或数据变更；事务完成后释放会话并返回响应。

1.1 三个核心组件的职责

中间件

中间件位于 HTTP 请求与路由处理函数之间，同时包裹响应返回过程。它适合处理横跨多个接口的通用逻辑，例如：

请求日志与响应日志
接口耗时统计
跨域处理
请求头检查
全局追踪标识
统一的安全策略

中间件关注的是一次 HTTP 请求的完整生命周期，而不是某个具体接口的业务细节。

依赖注入

依赖注入负责声明并提供路由所需要的资源。路由只描述“需要什么”，FastAPI 根据 Depends() 自动执行依赖函数并把结果传入路由。

常见依赖包括：

多个接口共享的分页参数
当前登录用户
权限校验结果
数据库会话
公共配置或服务对象

依赖注入能够把资源创建、参数校验、异常处理和资源释放从路由函数中分离出来。

SQLAlchemy 异步 ORM

ORM（Object-Relational Mapping，对象关系映射）在 Python 对象和关系型数据库表之间建立映射：

Python 类 Book       <──映射──>    数据库表 book
Book.id              <──映射──>    字段 id
Book.bookname        <──映射──>    字段 bookname
Book 实例            <──映射──>    表中的一行记录

SQLAlchemy 异步 API 使用 async/await 等待数据库 I/O。在等待数据库返回结果期间，事件循环可以处理其他可运行任务，从而提高 I/O 密集型服务的并发利用率。异步并不意味着单条 SQL 会执行得更快，其价值主要体现在等待期间不阻塞当前事件循环。

1.2 关键术语

术语	含义
`FastAPI`	基于 Python 类型提示构建 API 的 Web 框架
路由	HTTP 方法、URL 路径与处理函数之间的映射
中间件	包裹请求与响应过程的通用处理层
依赖项	由 FastAPI 解析、执行并注入路由的函数或可调用对象
`Depends`	声明路由依赖关系的工具
ORM	将数据库表、字段和记录映射为类、属性和对象
异步引擎	管理异步数据库连接与连接池的 `AsyncEngine`
会话	通过 `AsyncSession` 执行语句并管理 ORM 对象和事务
事务	作为一个整体提交或回滚的一组数据库操作
`DeclarativeBase`	SQLAlchemy 2.0 声明式 ORM 模型基类
`Mapped`	ORM 属性的类型标注容器
`mapped_column`	声明模型字段及其数据库约束
CRUD	Create、Read、Update、Delete，即新增、查询、更新、删除
标量值	单个值，例如总数、平均价格或单条 ORM 对象
结果集	数据库语句执行后返回的 `Result` 对象

1.3 示例代码的公共结构

异步 ORM 的基础设施通常由以下组成部分构成：

创建 FastAPI 应用；
创建 SQLAlchemy 异步引擎；
声明 Base 与 Book ORM 模型；
在应用启动时建表；
创建异步会话工厂；
通过依赖函数提供数据库会话；
在路由中执行不同的 CRUD 操作。

这些部分构成一套可复用的 ORM 基础设施；具体业务接口只需在此基础上编写不同的查询或数据变更逻辑。第二章先展示公共模板，再讨论各类操作的差异。

第二章中间件、依赖注入与异步 ORM

2.1 HTTP 中间件：包裹完整请求周期

FastAPI 使用 @app.middleware("http") 注册 HTTP 中间件。中间件函数接收两个核心参数：

request：当前请求对象；
call_next：将请求继续传递给下一处理层，并返回响应。

@app.middleware("http")
async def middleware2(request, call_next):
    print("中间件2 start")
    response = await call_next(request)
    print("中间件2 end")
    return response

@app.middleware("http")
async def middleware1(request, call_next):
    print("中间件1 start")
    response = await call_next(request)
    print("中间件1 end")
    return response

await call_next(request) 是执行顺序的分界点：

调用之前属于请求前置处理；
调用之后属于响应后置处理；
返回的 response 必须继续向外层返回。

在该示例的注册顺序下，中间件形成嵌套结构：

其结构与嵌套函数调用类似，后注册的中间件位于更外层。需要注意，中间件应保留全局、轻量的职责；只服务于个别路由的逻辑通常更适合放在依赖项或路由函数中。

2.2 依赖注入：公共参数与资源管理

新闻列表与用户列表都需要 skip 和 limit。若每个路由分别声明，校验规则和默认值容易产生差异。示例将公共参数封装为依赖：

async def common_parameters(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, le=60)
):
    return {"skip": skip, "limit": limit}

@app.get("/news/news_list")
async def get_news_list(commons=Depends(common_parameters)):
    return commons

@app.get("/user/user_list")
async def get_user_list(commons=Depends(common_parameters)):
    return commons

调用链可以表示为：

Depends(common_parameters) 声明的是依赖关系，而不是在定义路由时立即调用函数。FastAPI 会在请求期间解析该关系，并把依赖函数的返回值注入 commons。

依赖函数还可以使用 yield。yield 前负责创建资源，yield 后负责清理资源，因此非常适合数据库会话等具有生命周期的对象：

2.3 SQLAlchemy 异步 ORM 公共模板

以下模板整合了异步 ORM 中反复使用的数据库基础代码。为避免在每个 CRUD 操作中重复展示，后续只保留实际变化的语句。

from datetime import datetime

from fastapi import Depends, FastAPI
from sqlalchemy import DateTime, Float, String, func
from sqlalchemy.ext.asyncio import (
    AsyncSession,
    async_sessionmaker,
    create_async_engine,
)
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

app = FastAPI()

ASYNC_DATABASE_URL = (
    "mysql+aiomysql://root:123456@localhost:3306/"
    "FastAPI_first?charset=utf8"
)

async_engine = create_async_engine(
    ASYNC_DATABASE_URL,
    echo=True,
    pool_size=10,
    max_overflow=20,
)


class Base(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="修改时间",
    )


class Book(Base):
    __tablename__ = "book"

    id: Mapped[int] = mapped_column(primary_key=True, comment="书籍id")
    bookname: Mapped[str] = mapped_column(String(255), comment="书名")
    author: Mapped[str] = mapped_column(String(255), comment="作者")
    price: Mapped[float] = mapped_column(Float, comment="价格")
    publisher: Mapped[str] = mapped_column(String(255), comment="出版社")


async def create_tables():
    async with async_engine.begin() as conn:
        await conn.run_sync(Base.metadata.create_all)


@app.on_event("startup")
async def startup_event():
    await create_tables()


AsyncSessionLocal = async_sessionmaker(
    bind=async_engine,
    class_=AsyncSession,
    expire_on_commit=False,
)


async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

异步引擎与连接池

连接字符串：

mysql+aiomysql://root:123456@localhost:3306/FastAPI_first?charset=utf8
│     │            │    │         │
│     │            │    │         └─ 数据库名称与字符集
│     │            │    └────────── 主机与端口
│     │            └─────────────── 用户名与密码
│     └──────────────────────────── 异步驱动
└────────────────────────────────── 数据库方言

主要配置项如下：

参数	作用
`echo=True`	输出 SQL 日志，适合开发调试
`pool_size=10`	连接池常驻连接数量
`max_overflow=20`	连接池耗尽时允许临时增加的连接数量

连接池并不是为每次请求永久创建新连接，而是维护并复用一组连接。实际配置应结合数据库最大连接数、应用实例数量与请求并发量确定。

声明式模型

Book 类通过 __tablename__ = "book" 绑定数据库表。Mapped[T] 表示 Python 属性类型，mapped_column() 描述数据库字段约束。

Book.id: Mapped[int]
   │             │
   │             └─ Python 类型标注
   └─────────────── ORM 属性

mapped_column(primary_key=True)
                 │
                 └─ 数据库主键约束

Base 中的 create_time 和 update_time 会被继承到 Book 模型中，因此公共审计字段只需声明一次。

建表过程

async with async_engine.begin() as conn:
    await conn.run_sync(Base.metadata.create_all)

Base.metadata 收集所有继承自 Base 的表元数据；create_all() 根据元数据创建尚不存在的表。run_sync() 用于在异步连接中执行 SQLAlchemy 提供的同步 DDL 操作。

示例使用 @app.on_event("startup") 在启动时建表。该方式便于演示；在正式项目中，表结构变更通常交给 Alembic 等迁移工具管理，以保存版本历史并支持升级、回退。

会话工厂与事务边界

async_sessionmaker() 创建异步会话工厂。每次调用 AsyncSessionLocal() 都会生成一个独立的 AsyncSession。

async def get_database():
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise

该依赖定义了一条清晰的事务规则：

路由正常结束  ──> commit
路由发生异常  ──> rollback ──> 继续抛出异常
请求处理结束  ──> async with 自动关闭会话

expire_on_commit=False 表示提交后不立即使 ORM 对象中的属性过期，因此返回对象时通常不需要为已加载字段再次查询数据库。

2.4 路由中使用异步会话

@app.get("/book/books")
async def get_book_list(db: AsyncSession = Depends(get_database)):
    result = await db.execute(select(Book))
    books = result.scalars().all()
    return books

执行过程如下：

FastAPI 发现 Depends(get_database)；
get_database() 创建并产出 AsyncSession；
路由使用 db.execute() 异步执行查询；
scalars() 从结果行中提取 Book 对象；
all() 将全部对象收集为列表；
路由结束后，依赖函数继续执行事务处理并关闭会话。

这里需要区分三个对象：

对象	作用
`select(Book)`	SQL 表达式，描述需要执行的查询
`result`	数据库执行结果，可能包含多行、多列
`result.scalars().all()`	提取第一列中的 ORM 对象并形成列表

2.5 查询：从主键定位到聚合与分页

主键查询与普通查询

按主键获取单条记录时，AsyncSession.get() 最直接：

book = await db.get(Book, 1)

条件查询则通过 select() 和 where() 构造：

result = await db.execute(
    select(Book).where(Book.id == book_id)
)
book = result.scalar_one_or_none()

两种方式的侧重点不同：

db.get(Book, primary_key)：按照主键获取单个 ORM 对象；
select(Book).where(...)：支持任意查询条件、排序、关联、分页与组合表达式。

结果提取方法也对应不同语义：

方法	返回内容	典型用途
`scalar_one_or_none()`	一条记录或 `None`；多于一条会报错	唯一条件查询
`scalars().first()`	第一条记录或 `None`	只关心第一条
`scalars().all()`	ORM 对象列表	列表查询
`scalar()`	结果第一行第一列	聚合值或单个值

比较、模糊、逻辑与包含条件

# 等值条件
select(Book).where(Book.id == book_id)

# 范围条件
select(Book).where(Book.price >= 200)

# 模糊条件：作者以“曹”开头
select(Book).where(Book.author.like("曹%"))

# OR 条件
select(Book).where(
    (Book.author.like("曹%")) | (Book.price > 100)
)

# IN 条件
id_list = [1, 3, 5, 7]
select(Book).where(Book.id.in_(id_list))

LIKE 通配符的含义如下：

通配符	含义	示例
`%`	任意数量字符	`曹%` 匹配所有以“曹”开头的内容
`_`	单个字符	`曹_` 匹配“曹”后恰好一个字符

SQLAlchemy 表达式中的逻辑运算符为：

&：AND；
|：OR；
~：NOT。

组合条件时应使用括号明确优先级，避免 Python 运算符优先级导致表达式含义偏离预期。

聚合查询

count_stmt = select(func.count(Book.id))
max_stmt = select(func.max(Book.price))
sum_stmt = select(func.sum(Book.price))
avg_stmt = select(func.avg(Book.price))

result = await db.execute(avg_stmt)
average_price = result.scalar()

聚合函数返回单个计算值，因此通常使用 scalar() 提取。聚合查询不会返回完整 Book 对象，而是返回数量、最大值、总和或平均值等结果。

分页查询

skip = (page - 1) * page_size
stmt = select(Book).offset(skip).limit(page_size)
result = await db.execute(stmt)
books = result.scalars().all()

分页偏移量公式为：

offset = (page - 1) × page_size

例如 page=3、page_size=10 时，需要跳过前 20 条记录，再获取 10 条。

分页查询通常还应补充稳定排序：

stmt = (
    select(Book)
    .order_by(Book.id)
    .offset(skip)
    .limit(page_size)
)

没有明确的 order_by() 时，数据库不保证每次返回相同顺序，可能导致翻页时出现重复或遗漏。

2.6 新增、更新与删除

新增数据

请求体由 Pydantic 模型负责校验，随后转换为 ORM 对象并加入会话：

class BookCreate(BaseModel):
    id: int
    bookname: str
    author: str
    price: float
    publisher: str


@app.post("/book/add_book")
async def add_book(
    book: BookCreate,
    db: AsyncSession = Depends(get_database),
):
    book_obj = Book(**book.model_dump())
    db.add(book_obj)
    return book_obj

数据流向如下：

JSON 请求体
    ↓ Pydantic 校验
BookCreate 对象
    ↓ model_dump()
字典
    ↓ Book(**data)
ORM 对象
    ↓ db.add()
加入当前会话
    ↓ 依赖结束后 commit
写入数据库

原示例使用 book.__dict__ 并在路由中显式 commit()。在 Pydantic 2 中，model_dump() 是更明确的模型导出方式；如果事务已经由 get_database() 统一提交，路由通常不需要重复提交。

更新数据

更新采用“先查询，再修改”的对象化方式：

@app.put("/book/update_book/{book_id}")
async def update_book(
    book_id: int,
    data: BookUpdate,
    db: AsyncSession = Depends(get_database),
):
    db_book = await db.get(Book, book_id)
    if db_book is None:
        raise HTTPException(status_code=404, detail="查无此书")

    db_book.bookname = data.bookname
    db_book.author = data.author
    db_book.price = data.price
    db_book.publisher = data.publisher
    return db_book

被会话管理的 ORM 对象发生属性变化后，SQLAlchemy 会跟踪这些变化。事务提交时，对应的 UPDATE 语句会被发送到数据库。

删除数据

删除同样先确认目标记录存在：

@app.delete("/book/delete_book/{book_id}")
async def delete_book(
    book_id: int,
    db: AsyncSession = Depends(get_database),
):
    db_book = await db.get(Book, book_id)
    if db_book is None:
        raise HTTPException(status_code=404, detail="查无此书")

    await db.delete(db_book)
    return {"msg": "删除图书成功"}

路由正常结束后，由会话依赖统一提交删除事务。若查询不到记录，抛出 HTTPException(404)；异常会使依赖执行回滚，避免提交不完整的事务。

CRUD 规律汇总

操作	主要步骤	核心 API
新增	构造对象 → 加入会话 → 提交	`Book(...)`、`db.add()`
查询	构造语句 → 执行 → 提取结果	`select()`、`db.execute()`、`scalars()`
主键查询	指定模型和主键	`db.get()`
更新	查询对象 → 修改属性 → 提交	`db.get()`、属性赋值
删除	查询对象 → 标记删除 → 提交	`db.get()`、`db.delete()`

第三章总结

中间件、依赖注入与 SQLAlchemy 异步 ORM 分别对应 FastAPI 服务中的三个关键环节：请求处理、资源组织与数据持久化。

中间件包裹一次完整的 HTTP 请求与响应过程。call_next(request) 之前可执行请求前置逻辑，之后可执行响应后置逻辑，适合承载日志、耗时统计等通用能力。
依赖注入通过 Depends() 将公共参数、鉴权结果和数据库会话等资源按需提供给路由。依赖函数中的 yield 能够自然表达“创建资源 → 使用资源 → 清理资源”的生命周期。
SQLAlchemy 异步 ORM使用模型描述表结构，使用 AsyncSession 执行异步查询与 CRUD，并通过 commit() 和 rollback() 维护事务一致性。

一次典型请求的处理过程可以概括为：

请求进入
  → 中间件执行前置处理
  → FastAPI 解析路由与依赖
  → 路由使用 AsyncSession 执行数据库操作
  → 正常提交或异常回滚事务
  → 中间件执行后置处理
  → 返回响应

数据库操作可以归纳为四类固定模式：

操作	核心过程	常用 API
查询	构造语句 → 执行 → 提取结果	`select()`、`execute()`、`scalars()`
新增	构造 ORM 对象 → 加入会话 → 提交	`Book(...)`、`add()`、`commit()`
更新	查询目标 → 修改属性 → 提交	`get()`、属性赋值、`commit()`
删除	查询目标 → 删除对象 → 提交	`get()`、`delete()`、`commit()`