FastAPI 接入异步 PostgreSQL 完成任务 CRUD 与数据库迁移-开发者社区

内存列表写起来很轻松，服务一重启，昨天创建的任务就像没发生过。真正麻烦的还不只是丢数据，多个请求同时改一条任务时，列表也没有事务可言。这一篇把第一篇的接口换成 PostgreSQL，并让迁移脚本替我们记录表结构的变化。

配套代码已经放在 fastapi-task-api，文章中的完整实现以main分支为准。

让数据库连接成为配置

数据库地址不能散落在路由里。开发机、测试环境和 Docker 容器的主机名都不同，把它收进配置模型，部署时只需要替换环境变量。

frompydantic_settingsimportBaseSettings,SettingsConfigDictclassSettings(BaseSettings):database_url:str="postgresql+asyncpg://task_api:task_api@localhost:5432/task_api"model_config=SettingsConfigDict(env_file=".env")

项目使用 SQLAlchemy 2 的异步引擎和asyncpg驱动。异步不是让每条 SQL 更快，它让等待数据库返回的时间可以让给别的请求。

fromsqlalchemy.ext.asyncioimportAsyncSession,async_sessionmaker,create_async_engine engine=create_async_engine(settings.database_url,pool_pre_ping=True)SessionLocal=async_sessionmaker(engine,expire_on_commit=False,class_=AsyncSession)asyncdefget_session():asyncwithSessionLocal()assession:yieldsession# 一个请求拿到一个会话

pool_pre_ping=True会在复用连接前检查连接是否还活着。数据库重启后直接复用旧连接，是线上很常见的一类偶发错误。

模型描述表，Schema 描述接口

ORM 模型和 Pydantic 模型看起来字段相似，职责却不同。前者描述表、外键和索引，后者描述接口允许传入或返回什么。把两者硬合在一个类里，起步很快，后面加密码字段或内部状态时就会开始泄漏。

importuuidfromenumimportStrEnumfromsqlalchemyimportEnum,ForeignKey,Stringfromsqlalchemy.ormimportMapped,mapped_columnclassTaskStatus(StrEnum):TODO="todo"IN_PROGRESS="in_progress"DONE="done"classTask(Base):__tablename__="tasks"id:Mapped[uuid.UUID]=mapped_column(primary_key=True,default=uuid.uuid4)title:Mapped[str]=mapped_column(String(200))status:Mapped[TaskStatus]=mapped_column(Enum(TaskStatus),default=TaskStatus.TODO)owner_id:Mapped[uuid.UUID]=mapped_column(ForeignKey("users.id"),index=True)

这里已经预留了owner_id。第三篇才会引入用户认证，但表结构早点确定，迁移就不会反复推倒重来。

一次查询如何穿过依赖注入

路由不应该自己创建连接。Depends把会话传进函数，框架在请求结束后关闭它。回到任务列表这块，分页和状态筛选仍然是第一篇的接口，只是实现从切片换成 SQL。

fromsqlalchemyimportfunc,select@router.get("",response_model=TaskList)asyncdeflist_tasks(skip:int=Query(0,ge=0),limit:int=Query(20,ge=1,le=100),task_status:TaskStatus|None=Query(None,alias="status"),session:AsyncSession=Depends(get_session),)->TaskList:condition=Task.owner_id==current_user.idiftask_statusisnotNone:condition=condition&(Task.status==task_status)total=awaitsession.scalar(select(func.count()).select_from(Task).where(condition))rows=awaitsession.scalars(select(Task).where(condition).offset(skip).limit(limit))returnTaskList(items=list(rows),total=totalor0)

查询列表和统计总数是两条 SQL，这在多数后台页面足够清楚。数据量很大时再改用游标分页，不要为了一个十条数据的任务清单提前造复杂方案。

迁移不是可有可无的脚本

直接create_all()在本地很方便，团队协作就会变得危险。谁在什么时候加了列，没有可追溯记录。Alembic 把每次 schema 变更写成版本文件，发布时按顺序执行。

uv add alembic asyncpg sqlalchemy uv run alembic revision--autogenerate-m"create users and tasks"uv run alembic upgrade head

本项目的首个迁移创建users和tasks两张表，并为邮箱和任务所有者建立索引。自动生成的迁移也要人工审一遍，特别是删除列、枚举变化和大表加索引。工具只知道模型变了，不知道线上数据值不值得保留。

常见卡点

await少写一个，SQLAlchemy 往往不会立刻报出最直观的错误。session.execute、session.commit、session.refresh都是异步边界。另一个坑是把 ORM 对象原样返回，建议在响应模型上开启from_attributes=True，由 Pydantic 只挑选公开字段。

还有一点经常被忽略，commit后数据库生成的id和时间戳不会自动回到 Python 对象。创建接口里要await session.refresh(task)，否则响应有机会缺字段。