FastAPI 接入异步 PostgreSQL 完成任务 CRUD 与数据库迁移

内存列表写起来很轻松服务一重启昨天创建的任务就像没发生过。真正麻烦的还不只是丢数据多个请求同时改一条任务时列表也没有事务可言。这一篇把第一篇的接口换成 PostgreSQL并让迁移脚本替我们记录表结构的变化。配套代码已经放在 fastapi-task-api文章中的完整实现以main分支为准。让数据库连接成为配置数据库地址不能散落在路由里。开发机、测试环境和 Docker 容器的主机名都不同把它收进配置模型部署时只需要替换环境变量。frompydantic_settingsimportBaseSettings,SettingsConfigDictclassSettings(BaseSettings):database_url:strpostgresqlasyncpg://task_api:task_apilocalhost:5432/task_apimodel_configSettingsConfigDict(env_file.env)项目使用 SQLAlchemy 2 的异步引擎和asyncpg驱动。异步不是让每条 SQL 更快它让等待数据库返回的时间可以让给别的请求。fromsqlalchemy.ext.asyncioimportAsyncSession,async_sessionmaker,create_async_engine enginecreate_async_engine(settings.database_url,pool_pre_pingTrue)SessionLocalasync_sessionmaker(engine,expire_on_commitFalse,class_AsyncSession)asyncdefget_session():asyncwithSessionLocal()assession:yieldsession# 一个请求拿到一个会话pool_pre_pingTrue会在复用连接前检查连接是否还活着。数据库重启后直接复用旧连接是线上很常见的一类偶发错误。模型描述表Schema 描述接口ORM 模型和 Pydantic 模型看起来字段相似职责却不同。前者描述表、外键和索引后者描述接口允许传入或返回什么。把两者硬合在一个类里起步很快后面加密码字段或内部状态时就会开始泄漏。importuuidfromenumimportStrEnumfromsqlalchemyimportEnum,ForeignKey,Stringfromsqlalchemy.ormimportMapped,mapped_columnclassTaskStatus(StrEnum):TODOtodoIN_PROGRESSin_progressDONEdoneclassTask(Base):__tablename__tasksid:Mapped[uuid.UUID]mapped_column(primary_keyTrue,defaultuuid.uuid4)title:Mapped[str]mapped_column(String(200))status:Mapped[TaskStatus]mapped_column(Enum(TaskStatus),defaultTaskStatus.TODO)owner_id:Mapped[uuid.UUID]mapped_column(ForeignKey(users.id),indexTrue)这里已经预留了owner_id。第三篇才会引入用户认证但表结构早点确定迁移就不会反复推倒重来。一次查询如何穿过依赖注入路由不应该自己创建连接。Depends把会话传进函数框架在请求结束后关闭它。回到任务列表这块分页和状态筛选仍然是第一篇的接口只是实现从切片换成 SQL。fromsqlalchemyimportfunc,selectrouter.get(,response_modelTaskList)asyncdeflist_tasks(skip:intQuery(0,ge0),limit:intQuery(20,ge1,le100),task_status:TaskStatus|NoneQuery(None,aliasstatus),session:AsyncSessionDepends(get_session),)-TaskList:conditionTask.owner_idcurrent_user.idiftask_statusisnotNone:conditioncondition(Task.statustask_status)totalawaitsession.scalar(select(func.count()).select_from(Task).where(condition))rowsawaitsession.scalars(select(Task).where(condition).offset(skip).limit(limit))returnTaskList(itemslist(rows),totaltotalor0)查询列表和统计总数是两条 SQL这在多数后台页面足够清楚。数据量很大时再改用游标分页不要为了一个十条数据的任务清单提前造复杂方案。迁移不是可有可无的脚本直接create_all()在本地很方便团队协作就会变得危险。谁在什么时候加了列没有可追溯记录。Alembic 把每次 schema 变更写成版本文件发布时按顺序执行。uv add alembic asyncpg sqlalchemy uv run alembic revision--autogenerate-mcreate users and tasksuv run alembic upgrade head本项目的首个迁移创建users和tasks两张表并为邮箱和任务所有者建立索引。自动生成的迁移也要人工审一遍特别是删除列、枚举变化和大表加索引。工具只知道模型变了不知道线上数据值不值得保留。修改 ORM 模型生成迁移检查 SQL提交版本库部署执行 upgrade常见卡点await少写一个SQLAlchemy 往往不会立刻报出最直观的错误。session.execute、session.commit、session.refresh都是异步边界。另一个坑是把 ORM 对象原样返回建议在响应模型上开启from_attributesTrue由 Pydantic 只挑选公开字段。还有一点经常被忽略commit后数据库生成的id和时间戳不会自动回到 Python 对象。创建接口里要await session.refresh(task)否则响应有机会缺字段。数据库接入完成后任务终于能活过一次重启。但谁能读和改哪条任务还没有答案。下一篇把owner_id接到真实用户上再用测试把这些规则固定下来。本篇收口异步会话通过依赖注入按请求创建和释放ORM 管表结构Pydantic 管接口边界列表查询同时返回数据和总数支持分页与状态筛选Alembic 让 schema 变化有版本、可审查、可部署