开篇你是不是也半夜被报警吵醒过“叮——” 线上服务报错了sqlalchemy.exc.StatementError。原因可能就是顺手改了个模型字段名却忘了生成迁移脚本导致生产环境表结构对不上。这种经历我相信不少朋友都有过。尤其是从 FastAPI SQLAlchemy 起步的时候异步怎么配模型怎么写Alembic 为什么总是识别不到变更一个坑接一个坑。今天咱们就来聊聊从零到一搭一套靠谱的数据库层顺便把我踩过的坑都给你标上警示牌。 先理一下咱们要干的事 第一部分为什么是这套组合聊聊选型逻辑 第二部分手把手搭环境从配数据库到写第一个模型 第三部分Alembic 接入与避坑自动生成脚本的那些坑 第四部分进阶思考连接池、异步、以及工程化建议️ 一、为什么是 FastAPI SQLAlchemy 2.0 Alembic咱们可以把 API 比作一家餐厅FastAPI 是那个手脚麻利的点餐员能快速把客人的需求HTTP请求传给后厨SQLAlchemy 就是后厨的食材管理员负责管理所有食材数据的进出和记录而Alembic 则是食材管理员的变更日志本每次新增食材或调整存储方式都得在本子上记一笔保证后厨和仓库一致。SQLAlchemy 2.0 之后语法更清爽了但同时也带来了一些变化——比如必须用 Mapped 和 mapped_column如果你还抱着 1.x 的写法跑起来就会报错。别问我怎么知道的我第一次升级项目时整个 models.py 一片飘红。️ 二、从零开始搭环境附代码可复制 安装依赖先装好这些包注意版本pip install fastapi uvicorn sqlalchemy alembic asyncpg # asyncpg 是 PostgreSQL 异步驱动这里多说一句如果你用 MySQL请装 aiomysql 或 asyncmy千万别装错否则异步引擎跑不起来。️ 项目结构myproject/ ├── app/ │ ├── __init__.py │ ├── database.py # 引擎、会话配置 │ ├── models.py # SQLAlchemy 模型 │ └── main.py # FastAPI 应用 ├── alembic.ini └── alembic/ # 迁移目录 配置数据库连接database.py这是最容易踩坑的地方。SQLAlchemy 2.0 推荐使用异步方式但很多人照抄旧代码配了个同步引擎然后和 FastAPI 的异步路由打架。from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession from sqlalchemy.orm import DeclarativeBase DATABASE_URL postgresqlasyncpg://user:passlocalhost/dbname # 创建异步引擎 engine create_async_engine(DATABASE_URL, echoTrue) # echoTrue 会打印SQL开发时很有用 # 创建异步会话工厂 AsyncSessionLocal async_sessionmaker(engine, expire_on_commitFalse) # 基类所有模型都要继承它 class Base(DeclarativeBase): pass注意看我用了create_async_engine和async_sessionmaker这才是异步的正确姿势。echoTrue 在开发时打开能让你看到背后执行的 SQL但线上记得关掉否则日志会爆炸。 定义模型models.pySQLAlchemy 2.0 的新写法用Mapped和mapped_column替代老旧的 Column。刚开始我很不习惯但用顺手后发现类型提示更香了。from sqlalchemy.orm import Mapped, mapped_column from app.database import Base class User(Base): __tablename__ users id: Mapped[int] mapped_column(primary_keyTrue) name: Mapped[str] mapped_column(nullableFalse) email: Mapped[str] mapped_column(uniqueTrue, indexTrue) age: Mapped[int | None] # 允许为空的字段用 Optional 或 | None注意如果字段允许为空类型注解必须用 | None否则 mapped_column 会默认 nullableFalse。我一开始就漏了这个导致插入数据时一直报错。⚡ 在 FastAPI 中使用异步会话接下来咱们在 main.py 里写一个依赖每次请求生成独立的会话from fastapi import FastAPI, Depends from sqlalchemy.ext.asyncio import AsyncSession from app.database import AsyncSessionLocal from app import models app FastAPI() async def get_db() - AsyncSession: async with AsyncSessionLocal() as session: yield session app.get(/users) async def get_users(db: AsyncSession Depends(get_db)): result await db.execute(select(models.User)) users result.scalars().all() return users这里用了async with自动管理会话生命周期用完即关不会泄露连接。这是官方推荐的做法比手动 close 靠谱多了。 三、Alembic 接入与自动迁移现在模型有了但数据库里还没表呢。这时候 Alembic 就该登场了。很多人觉得 Alembic 麻烦但其实只要配一次后面爽到飞起。⚙️ 初始化 Alembicalembic init alembic这会在项目根目录生成 alembic.ini 和 alembic/ 文件夹。然后打开 alembic.ini修改数据库连接字符串sqlalchemy.url postgresqlasyncpg://user:passlocalhost/dbname注意这里必须用异步驱动格式否则后面生成迁移时会报错。 修改 env.py 支持异步这是最关键的一步默认生成的 env.py 是给同步用的我们需要改成异步模式。打开 alembic/env.py找到 run_migrations_online 函数修改如下from sqlalchemy.ext.asyncio import async_engine_from_config from sqlalchemy import pool import asyncio # 在 run_migrations_online 内部 def do_run_migrations(connection): context.configure(connectionconnection, target_metadatatarget_metadata) with context.begin_transaction(): context.run_migrations() async def run_async_migrations(): connectable async_engine_from_config( config.get_section(config.config_ini_section), prefixsqlalchemy., poolclasspool.NullPool, ) async with connectable.connect() as connection: await connection.run_sync(do_run_migrations) asyncio.run(run_async_migrations())这段代码的作用是让 Alembic 用异步引擎连接数据库然后通过run_sync调用同步的迁移配置。如果不改这里你执行迁移时会得到 “asyncpg can‘t run in sync” 之类的错误。这个坑我替你们填了直接复制就行。还有一处要更改否则会报错 env.py does not provide a MetaData object or sequence of objects to the contextfrom app.models import User # target_metadata None target_metadata User.metadata 生成并执行迁移alembic revision --autogenerate -m init user table alembic upgrade head运行第一条命令后去 alembic/versions/ 下检查生成的脚本。千万别直接相信自动生成的脚本有时候它可能漏掉索引或者把 nullable 搞反。我每次都会手动 review 一遍比如看有没有创建唯一约束有没有遗漏外键。这是线上事故的高发区。确认无误后执行 upgrade表就建好了。以后每次修改模型重复这两步即可。 四、进阶思考与注意事项 同步与异步的抉择如果你是新项目强烈建议用异步。但如果你维护的老项目是同步的也别急着重构。同步配合fastapi 的 run_in_threadpool也能用只是性能上限低一些。不过根据我的线上经验大部分业务场景异步的收益并没有想象中大除非你是高并发 IO 密集型。 连接池调优默认的连接池参数可能不适合你的并发量。在 create_async_engine 中可以调整engine create_async_engine(DATABASE_URL, pool_size20, max_overflow10)pool_size是核心连接数max_overflow是峰值时最多可以创建的额外连接数。