FastAPI数据层实战：异步SQLAlchemy、CRUD与事务控制全解析-平芜编程栈

很多朋友问到FastAPI的数据操作到底怎么写才是规范、能扛住生产环境的，这期教程正好填上这个坑。前六篇我们把FastAPI的接口定义、参数校验、依赖注入和中间件都过了一遍，但很多项目走到数据库这一层就卡住了——SQLAlchemy的同步写法放进async接口里容易阻塞事件循环，ORM模型和Pydantic模型来回转换容易绕晕，事务一多就不知道什么时候commit、什么时候rollback。

这篇教程就把数据操作这一整条链路讲透：从异步引擎配置、模型设计、CRUD接口，到批量导入导出和事务控制，全部给出可以直接抄进项目里的写法。内容偏向实战，会结合我实际项目里踩过的坑来说明，适合已经会写基础FastAPI接口、想正经做数据层的开发者参考，也适合正在准备FastAPI相关面试的人用来补全知识点。

1. 数据操作的整体设计与方案选型

1.1 为什么数据层必须单独设计

FastAPI本身是不关心数据操作的，它只负责接收请求、调用你提供的函数、把返回值序列化成JSON。所以“数据操作”这部分的质量，完全取决于你在FastAPI下面怎么组织数据库访问层。很多初学者习惯在路由函数里直接写SQLAlchemy查询，项目小的时候没问题，一旦路由变多、逻辑变复杂，查询代码会散落在各个接口里，之后想统一加缓存、统一处理事务、统一做分页都会非常痛苦。

所以我建议在项目里把数据访问单独拆出一层，大致目录结构是这样：

app/ ├── main.py # 应用入口，包含路由注册与启动逻辑 ├── core/ │ ├── config.py # 环境变量与全局配置 │ └── database.py # 引擎、会话工厂、session依赖 ├── models/ # SQLAlchemy ORM模型 │ ├── __init__.py │ └── product.py ├── schemas/ # Pydantic模型，用于接口入参与出参 │ ├── __init__.py │ └── product.py ├── crud/ # 数据操作层，每个模型对应一个文件 │ ├── __init__.py │ └── product.py ├── api/ │ └── v1/ │ ├── __init__.py │ └── product.py # 路由层，只负责HTTP相关处理 └── alembic/ # 数据库迁移脚本目录

这种分层方式的好处是：路由层不碰数据库，只调用crud层函数；crud层不依赖Request和Response对象，只处理数据；模型层不掺入业务逻辑。各层职责清晰，后续上单元测试也会很容易——直接测crud层，不需要启动HTTP服务。

有一点必须说明：这个结构不是FastAPI规定的，是我从不同项目里总结出来的通用做法。如果项目特别小，几个人临时合作写个脚本，你完全可以简化，但只要是预期能上线、会持续迭代的工程，建议一开始就按这个来。中途拆分比一开始拆分痛苦得多。

1.2 同步还是异步：选型的关键判断依据

FastAPI可以同时跑同步和异步接口，但在数据库层面选型时，需要先想清楚你的服务到底要面对什么样的并发场景。如果你的接口里只有非常简单的查询，而且预估并发不高，用同步SQLAlchemy加上线程池也能撑住；但如果你做的是面向大量小请求的业务系统，比如电商C端接口、消息推送服务，那么事件循环被数据库阻塞一次，整个进程能处理的请求数就会直线下降。

我这边项目最终选择了异步SQLAlchemy，具体来说是SQLAlchemy 2.0的异步特性配合asyncpg驱动连接PostgreSQL。异步写法和同步写法在模型定义上基本一致，主要区别在查询执行时要用await session.execute(...)。实测下来，在同样4核8G的容器里，纯查询接口的QPS比同步方案高出将近一倍，有IO等待的场景提升更明显。

为什么选2.0版本而不是1.4？因为2.0把传统的Query API真正淘汰了，统一换成select()、update()、delete()这种2.0风格的语句。虽然1.4也有兼容模式，但已经接触新项目的朋友，没必要再学一套即将过时的写法。

当然不是说所有项目都无脑上异步。如果你的业务大量涉及复杂报表、多表嵌套子查询、SQL调优，异步带来的复杂度可能大于收益。这种场景下，我反而建议用同步SQLAlchemy配合FastAPI的线程池，或者更干脆一点，直接用上文件里封装好的查询函数，别让业务代码直接碰SQLAlchemy。把“选型”这个问题想明白了，后面代码写起来会顺很多。

2. 数据库连接与会话管理

2.1 从配置文件到异步引擎

所有配置都应该从环境变量读取，不写死在代码里。用pydantic-settings管理配置是目前比较常见的做法，它能自动读取.env文件，并且在启动时做类型校验。下面这是config.py的快速实现：

from pydantic_settings import BaseSettings class Settings(BaseSettings): database_url: str = "postgresql+asyncpg://user:pass@localhost:5432/app_db" echo_sql: bool = False pool_size: int = 10 max_overflow: int = 20 pool_recycle: int = 3600 class Config: env_file = ".env" settings = Settings()

database_url里我用的是PostgreSQL的异步驱动asyncpg，这也是生产环境最常用的组合。很多人会问为什么不用同步驱动psycopg2，原因很简单：asyncpg在异步场景下性能和协议支持都更好，SQLAlchemy对它的支持也很成熟。

接下来是database.py，负责创建引擎和会话工厂：

from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from sqlalchemy.orm import DeclarativeBase from core.config import settings engine = create_async_engine( settings.database_url, echo=settings.echo_sql, pool_size=settings.pool_size, max_overflow=settings.max_overflow, pool_recycle=settings.pool_recycle, ) SessionLocal = async_sessionmaker( bind=engine, class_=AsyncSession, expire_on_commit=False, ) class Base(DeclarativeBase): pass

这里有个容易被忽略的参数：expire_on_commit=False。默认情况下，commit之后所有实例上的属性会失效，下一次访问会重新触发SQL查询。在异步环境里，这可能会在response序列化时意外触发懒加载，然后抛MissingGreenlet异常。把它设为False，commit后对象属性仍然可读，接口返回时会省很多麻烦。

pool_size和max_overflow要与实际部署环境匹配。我给的是一个相对保守的起步配置，总连接数约等于pool_size加max_overflow。这个值不能拍脑袋定，我一般按“单个实例能够同时处理的最大并发数据库操作数”来估算。比如你的服务实例有4个worker，每个worker并发跑20个请求，那连接池理论上至少要80个连接，否则请求会排队等待数据库连接。当然也不能无限调大，数据库服务端有最大连接数限制，连接太多反而会拖垮数据库。

2.2 Session生命周期与FastAPI依赖注入

Session的正确管理方式是：每个请求创建一次，请求结束关闭。不要全局复用一个session——这是新手最容易犯的错误。全局session在并发请求下会出现状态串线、事务相互干扰的问题。

在FastAPI里做这件事非常顺手，用依赖注入就行。我们在database.py里再定义一个生成器函数：

async def get_session() -> AsyncSession: async with SessionLocal() as session: yield session

然后在路由里这样使用：

from fastapi import APIRouter, Depends from sqlalchemy.ext.asyncio import AsyncSession from core.database import get_session from crud import product as product_crud router = APIRouter() @router.get("/products/{product_id}") async def get_product(product_id: int, db: AsyncSession = Depends(get_session)): product = await product_crud.get_product(db, product_id) return product

这里yield之前的代码在创建session，yield之后的代码在关闭session。async with块确保即使接口内部抛出异常，session也会正常关闭，连接归还给连接池。

有一点要特别注意：依赖注入里同一个session会在一个请求的整个生命周期中复用。这意味着如果你在接口里手动执行了session.commit()，之后又继续操作数据库，这些操作会开启新的事务，和之前的事务并没有关联。因此我建议接口层面不要直接调commit，而是让crud层里的写操作函数自己控制事务，或者在crud层之外再包一层service逻辑，统一提交。这块在后续事务部分会详细说。

3. ORM模型设计与数据表映射

3.1 类型选择和字段约束

模型设计直接决定数据操作的复杂程度。用SQLAlchemy 2.0的Mapped写法定义一个商品表，示例模型如下：

from datetime import datetime, timezone from typing import Optional from sqlalchemy import String, Integer, Numeric, DateTime, Text from sqlalchemy.orm import Mapped, mapped_column from core.database import Base class Product(Base): __tablename__ = "products" id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) name: Mapped[str] = mapped_column(String(128), nullable=False, unique=True, index=True) description: Mapped[Optional[str]] = mapped_column(Text, default=None) price: Mapped[float] = mapped_column(Numeric(10, 2), nullable=False) stock: Mapped[int] = mapped_column(Integer, nullable=False, default=0) created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now()) updated_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())

价格字段必须用Numeric而不是Float。Float在数据库里存储的是二进制浮点数，做金额计算会出现莫名的小数误差；Numeric则是精确数值类型。这在涉及金额的项目里是常识，但还是会有人在这里踩坑。

unique=True和index=True可以组合使用。unique本身会创建索引，单独设置index是为了让其他查询条件也能走索引。比如name字段加了unique约束，按name精确查询已经能走索引了；但如果经常按价格范围查、按创建时间排序，这些字段就应该加index。

created_at使用server_default=func.now()，让数据库来生成时间，而不是我们在Python代码里手动赋值。好处是多个应用实例同时写入时，时间由数据库统一生成，不会因为机器时钟偏差导致排序混乱。updated_at用onupdate=func.now()，每次做UPDATE操作时数据库会自动刷新这个字段，我们不需要在crud代码里维护它。

3.2 Alembic迁移与表结构同步

模型定义好之后，需要把表结构同步到数据库。开发环境可以简单调用Base.metadata.create_all()，但生产环境不建议这么做。原因是：create_all只会创建缺失的表，不会修改已经存在的表结构。项目迭代过程中要加字段、改类型，create_all完全帮不上忙。迁移工具可以记录每一次结构变更，并且支持升降级，团队协作时其他人拉代码后也能快速把本地库迁移到最新的schema。

我通常用Alembic配合异步引擎做迁移。初始化之后，先改数据库连接配置指向async的URL，再写迁移脚本。生成迁移的命令大致是：

alembic revision --autogenerate -m "create products table" alembic upgrade head

自动生成的迁移文件会包含create_table、add_column之类的操作。我建议每次提交迁移前都把生成的脚本打开检查一遍，因为autogenerate不是万能的，索引名变更、约束调整、数据迁移这类操作经常需要手写补充。另外，执行迁移时连接的是哪个库要在alembic配置里看仔细，我吃过一次亏，把测试环境的配置发到生产库上执行，好在当时只是建了个空表，没有造成数据损失。后续我把所有环境配置都改成了独立的.env文件，并加上了迁移前的提示确认，才彻底避免这类事故。

4. 核心CRUD接口实操

4.1 创建数据：模型实例构造与错误处理

写一条商品数据，crud层里可以这样封装：

from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from models.product import Product from schemas.product import ProductCreate async def create_product(db: AsyncSession, data: ProductCreate) -> Product: product = Product( name=data.name, description=data.description, price=data.price, stock=data.stock, ) db.add(product) await db.commit() await db.refresh(product) return product

commit之后必须refresh一次。因为id、created_at、updated_at这些字段是由数据库生成的，commit之后ORM实例上并没有这些值，refresh会重新从数据库加载这些字段。如果不做refresh，返回给前端的产品数据里id就是None，前端往后拿这个id去请求详情就全部404了。

这里的建议是：create_product的errors处理放在接口层做。比如数据库里name有唯一约束，插入重复名称会抛IntegrityError，接口层可以捕获后返回409冲突。crud层尽量保持纯净，不要混杂HTTP细节。

4.2 查询数据：主键、列表与分页

单条查询很简单，get和select两种写法都可以：

async def get_product(db: AsyncSession, product_id: int) -> Optional[Product]: return await db.get(Product, product_id)

列表加分页正确写法如下：

from sqlalchemy import select, func async def list_products( db: AsyncSession, page: int = 1, page_size: int = 20, keyword: str | None = None, ): stmt = select(Product) if keyword: stmt = stmt.where(Product.name.ilike(f"%{keyword}%")) total = await db.scalar(select(func.count()).select_from(stmt.subquery())) stmt = stmt.order_by(Product.id.desc()).offset((page - 1) * page_size).limit(page_size) rows = (await db.execute(stmt)).scalars().all() return rows, total

count子查询用了subquery而不是直接在原stmt上拼select(func.count())。因为原stmt可能包含order_by，直接在它基础上加count会把排序也带进去，虽然结果正确，但会多一次无谓的排序开销。先包一层subquery，再去掉排序逻辑，效率会好一些。

分页方式这里用offset/limit，适用于中小规模数据。如果表数据量达到百万级，更推荐keyset分页（也叫cursor分页），用上次返回的最后一条id作为位置标记。keyset分页不会随着页数增大而变慢，避免了offset大页面时数据库全表扫描的问题。两者没有绝对好坏，看数据量级和场景选择。

4.3 更新与删除：注意行级影响

更新操作有两种方案：一种是查出来再改属性，一种是用update语句直接执行。第一种适合更新前需要校验业务规则的场景，比如库存变更前要确认当前值；第二种适合无条件的字段更新，效率更高。同时更新多个字段时，用update语句：

from sqlalchemy import update async def update_product( db: AsyncSession, product_id: int, data: ProductUpdate, ) -> Optional[Product]: values = data.model_dump(exclude_unset=True) if not values: return await get_product(db, product_id) stmt = ( update(Product) .where(Product.id == product_id) .values(**values) .returning(Product) ) result = await db.execute(stmt) await db.commit() return result.scalar_one_or_none()

exclude_unset=True表示只提交前端传了字段。比如前端只想改stock，不传name，name就不会被覆盖。这是PATCH语义的正确实现。

删除操作同样可以用delete语句，但要注意外键依赖。如果子表里有数据引用这个产品，直接删除会抛外键约束错误。我通常会把删除做成软删除——加一个is_deleted字段，查询时默认过滤掉已删除的数据。这样做的好处是数据可回溯，也避免外键级联问题。到底用物理删除还是软删除，取决于业务需求，但作为一种数据操作方案，我建议业务系统优先考虑软删除。

5. 批量导入导出设计与事务处理

5.1 商品批量导入：从上传到入库的完整链路

在实际项目中，管理后台经常需要批量录入数据。与其一条条调用创建接口，不如设计一个导入接口：前端上传CSV或Excel文件，后端解析、校验、批量写入。这个场景在电商后台尤其常见——供应商给的商品价格表、库存表，动辄几千行，逐条调用接口既不高效，也容易把数据库和网络打满。

导入接口的实现思路是：

接收上传文件，保存为临时文件（或直接读入内存）。
用pandas或openpyxl解析文件内容。
逐行校验数据：名称非空、价格大于0、库存非负、是否存在重复。
将有效数据批量入库，无效数据收集起来返回给前端，方便用户下载错误报告。

批量入库这一步值得展开。很多人会自然地写一个循环，每条数据add一次，最后commit。这个做法能用，但性能很差。几千行的循环会导致几千次SQL flush，耗时会放大很多倍。更合适的做法是用bulk操作：

from sqlalchemy.dialects.postgresql import insert as pg_insert async def bulk_upsert_products(db: AsyncSession, rows: list[dict]): stmt = pg_insert(Product).values(rows) stmt = stmt.on_conflict_do_update( index_elements=[Product.name], set_={ "price": stmt.excluded.price, "stock": stmt.excluded.stock, "description": stmt.excluded.description, }, ) await db.execute(stmt) await db.commit()

这段代码用的是PostgreSQL的upsert语法：按name判断是否已存在，存在则更新价格、库存、描述，不存在则插入。一条SQL解决几千行数据，效率远高于循环逐条操作。

有个性能经验值得分享：批量入库时，每批建议控制在500到1000行。太少起不到批量的效果，太多则单条SQL过长，可能触发数据库的SQL语句大小限制或参数数量限制。如果是几万行数据，分批次执行，不要一把梭。

5.2 商品批量导出：流式生成保护内存

导出正好和导入相反：数据量大时不要一次性把所有数据都加载到内存里。比如导出10万商品为CSV，如果先全部查询出来再序列化，应用进程内存可能直接翻几倍。正确做法是用流式响应，分页读取数据，逐行写入响应流。

FastAPI可以用StreamingResponse实现这一点：

from fastapi.responses import StreamingResponse import csv, io async def export_products(db: AsyncSession, page_size: int = 1000): def generate(): yield "id,name,price,stock\n" last_id = 0 while True: stmt = ( select(Product) .where(Product.id > last_id) .order_by(Product.id) .limit(page_size) ) rows = (await db.execute(stmt)).scalars().all() if not rows: break for p in rows: yield f"{p.id},{p.name},{p.price},{p.stock}\n" last_id = rows[-1].id return StreamingResponse( generate(), media_type="text/csv", headers={"Content-Disposition": "attachment; filename=products.csv"}, )

这段代码用keyset分页方式替代offset，避免深分页性能问题；每次只取1000条，处理完立即释放，内存占用非常稳定。

需要指出的是，这里生成器函数里用了异步查询，所以generate本身要能配合事件循环运行。实际项目中我一般会把查询逻辑抽到crud层，让生成器函数只做拼装输出，这样结构更清晰，测试也更方便。

5.3 事务边界与并发控制

事务是数据操作里最核心的问题之一，也是面试官最喜欢深挖的话题。一个事务要保证的是：要么全部成功，要么全部失败，不能出现“库存扣了但订单没创建”这种中间状态。

FastAPI里管理事务有两个层级。小范围用session.begin()块即可：

async with SessionLocal() as session: async with session.begin(): await session.execute(update_sql_1) await session.execute(update_sql_2)

begin块结束时会自动commit，块内任何一步抛异常都会自动rollback，不需要手动编写commit/rollback逻辑。这是我在crud层写复杂写操作时最常用的方式。

大范围跨多个crud函数的事务，建议在service层组合。比如“下单”需要同时减库存和生成订单，可以把两个crud函数放在同一个session里，由service决定提交时机：

async def create_order(db: AsyncSession, user_id: int, items: list[dict]): async with db.begin(): order = await order_crud.create_order(db, user_id) for item in items: await stock_crud.deduct_stock(db, item.product_id, item.quantity) await order_item_crud.create_order_item(db, order.id, item) return order

这里db.begin()是外层事务，内层crud函数里不能再有独立的commit。这是一个约定——写crud函数时，保持“只写SQL，不发言提交”，由调用方控制事务边界。团队协作时最好把这条写进开发规范，否则很容易出现两边都在控制事务，导致保存点上下文错乱。

并发控制这块，最实用的手段是版本号乐观锁。在模型里加一个version字段：

version: Mapped[int] = mapped_column(Integer, nullable=False, default=0)

在SQLAlchemy的__mapper_args__中配置version_id_col指向它：

__mapper_args__ = {"version_id_col": version}

这样每次UPDATE时，SQLAlchemy会带上WHERE version = 当前值，更新成功后version加1。如果两个请求同时读到version=1并尝试更新，只有第一个能成功，第二个update会影响到0行记录，从而可以判断发生了并发冲突。这种方案很适合高并发扣库存、改配置这种场景。

6. 常见问题与排查技巧实录

6.1 MissingGreenlet错误：异步懒加载的坑

async环境里最常见的坑就是MissingGreenlet。触发原因通常是：ORM实例在session关闭之后被访问了还未加载的关联属性，或者懒加载触发时SQLAlchemy试图在隐式IO中执行查询，但异步环境不允许这么做。

它的典型报错长这样：

sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place?

发生场景一般是：接口从数据库查到一条订单记录，在session关闭后，业务代码访问order.items这个关联属性。默认情况下items是懒加载的，要等到访问时才去查数据库，但此时session已经关了，于是报错。

解决办法是查询时主动声明加载关联关系，用selectinload：

from sqlalchemy.orm import selectinload stmt = select(Order).options(selectinload(Order.items)).where(Order.id == order_id)

selectinload会用一条额外的IN查询把关联数据一次性加载，避免懒加载。注意selectinload适合集合关联（一对多、多对多），单对象关联（多对一）用joinedload更合适。

这个坑排查起来会稍微费时间，因为报错不一定马上出现，和请求的并发时序有关。我建议把“查询必带selectinload”写成crud层的代码规范，保证所有会返回给前端的数据都提前加载好，不依赖懒加载。

6.2 连接池耗尽与超时

接口突然大面积502或超时，排查时发现数据库连接池被占满，这是生产环境很棘手的问题之一。

最典型的诱因：session没有正确关闭。每个请求占用的连接都被高并发请求长时间持有，数据库连接数被逐渐耗尽。排查思路是打印连接池使用状态，或者直接查看数据库侧的活跃连接来源。修复方式是保证get_session依赖里用async with管理session，确保请求结束、异常抛出时session都会关闭。

第二个诱因是接口里的慢查询。某个查询没走索引，执行时间几秒，导致连接被长时间占用。排查方式是在SQLAlchemy配置里开启echo_sql，或者更轻量的方式是记录慢查询日志，定位后再给查询条件加合适的索引或改写查询逻辑。

第三个诱因是连接池配置过小。如果代码没问题、慢查询也不多，但并发就是上不去，那就回到2.1节里的连接池配置，把pool_size和max_overflow调大，同时检查数据库服务端的max_connections。

6.3 序列化递归与循环引用

在写出参时，Pydantic模型如果嵌套了父对象和子对象，很容易写出来一个相互引用的结构，导致序列化时无限递归，最后栈溢出。

解决方法是明确写出只向外序列化一层，避免在Pydantic里把双向关系都暴露出来。比如Order模型里有items，OrderItem模型里不要再嵌套Order，只需要order_id。也就是说，ORM模型之间可以双向关联，但Pydantic模型必须设计成单向的树形结构。这个原则几乎所有序列化框架都适用。

如果遇到ORM和Pydantic字段不完全匹配的情况，比如ORM里是snake_case，前端要camelCase，可以写一个转换函数手动映射字段，不要硬靠框架魔改。保持映射逻辑显式化，代码反而更容易维护。

6.4 查询N+1问题

N+1问题的表现：查询10条订单，每条订单又各查一次关联表，共执行了1次主查询加10次关联查询。接口响应时间随数据量线性恶化。

排查方法是在SQLAlchemy的echo_sql日志里观察SQL条数。如果发现查询列表后跟着大量相同结构的SELECT，那基本就是N+1了。修复方法就是6.1节说的selectinload或joinedload，在查询时一次性把关联数据加载出来。

还有一个优化方向是查询时只select需要的列，不select整个ORM对象。比如列表页只需要id、name、price，就可以用select(Product.id, Product.name, Product.price)返回Row而不是完整ORM实例。这样数据库传输的数据量小，ORM实例化和追踪的开销也少。但这个优化会让代码多写几行，不是性能瓶颈的项目不需要过度优化。

7. 数据操作的一些个人体会

从我自己的项目经验来看，数据操作层的核心不是把CRUD写出来，而是把边界划清楚：查询只管查询，事务由调用方控制，模型不掺业务逻辑，Pydantic模型单向序列化。这四条规矩守住了，无论项目怎么膨胀，数据层都不会烂到不可维护。

另外多说一句关于“抄作业”的建议：网上有很多FastAPI项目的仓储代码，GitHub上各种clean architecture模板也很多。我的建议是，与其直接复制一个庞大的模板，不如先理解template里每一层是做什么的，然后根据自己项目的实际复杂度做取舍。手里有五六条路的时候，别全铺上，选最简单的、但能满足未来三个月扩展需求的那条就好。

这套数据操作的方案在我这边的项目里已经跑了一年多，上过生产、扛过促销峰值，目前没有遇到新的绕不过去的坎。如果你按这个思路搭完数据层，遇到什么奇怪的报错或者有其他更好的方案，欢迎一起交流。