写接口一时爽，改模型火葬场？别怕，这篇手把手带你避开所有暗礁。
📌 摘要
本文以实战为导向，从零开始搭建 FastAPI + SQLAlchemy 2.0 + Alembic 的数据库层。通过生活化比喻和踩坑案例，讲透异步引擎配置、模型定义新写法、迁移脚本生成与审核等核心环节，帮你建立一套可靠、可维护的数据库操作实践。适合所有被数据库折腾过的后端开发者。
🎯 开篇：你是不是也半夜被报警吵醒过？
“叮——” 线上服务报错了：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 = "postgresql+asyncpg://user:pass@localhost/dbname"
# 创建异步引擎
engine = create_async_engine(DATABASE_URL, echo=True) # echo=True 会打印SQL，开发时很有用
# 创建异步会话工厂
AsyncSessionLocal = async_sessionmaker(engine, expire_on_commit=False)
# 基类，所有模型都要继承它
class Base(DeclarativeBase):
pass
注意看，我用了 create_async_engine 和 async_sessionmaker，这才是异步的正确姿势。echo=True 在开发时打开，能让你看到背后执行的 SQL，但线上记得关掉，否则日志会爆炸。
🧱 定义模型（models.py）
SQLAlchemy 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_key=True)
name: Mapped[str] = mapped_column(nullable=False)
email: Mapped[str] = mapped_column(unique=True, index=True)
age: Mapped[int | None] # 允许为空的字段，用 Optional 或 | None
注意：如果字段允许为空，类型注解必须用 | None，否则 mapped_column 会默认 nullable=False。我一开始就漏了这个，导致插入数据时一直报错。
⚡ 在 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 麻烦，但其实只要配一次，后面爽到飞起。
⚙️ 初始化 Alembic
alembic init alembic
这会在项目根目录生成 alembic.ini 和 alembic/ 文件夹。然后打开 alembic.ini，修改数据库连接字符串：
sqlalchemy.url = postgresql+asyncpg://user:pass@localhost/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(connection=connection, target_metadata=target_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),
prefix="sqlalchemy.",
poolclass=pool.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 context
from 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_size=20, max_overflow=10)
pool_size 是核心连接数，max_overflow 是峰值时最多可以创建的额外连接数。
我通常会根据服务器的数据库最大连接数来设置，比如 MySQL 默认 151，那么 pool_size+max_overflow 最好不要超过 150，留一点给管理工具。
🚨 永远不要在迁移脚本中直接改数据
Alembic 是用来改表结构的，不是用来改数据的。如果你需要数据迁移（比如把用户名从两列合并成一列），请在 upgrade 函数里用 op.execute() 执行原生 SQL，并且务必写好 downgrade 回退脚本。
血的教训：有一次我直接在脚本里写了 Python 循环更新数据，结果生产执行了半小时，锁表锁到崩溃。
👋 最后啰嗦一句
好了，这套组合拳打下来，你应该能顺利搭出一个稳当的数据库层了。其实这些工具都很成熟，难的是那些细枝末节的配置和习惯。我今天分享的每一个坑，都是真金白银换来的，希望你能绕过。
如果你觉得这篇文章有用，点赞、收藏、关注 走一波，下次找出来也方便。还有什么数据库相关的问题，欢迎留言，咱们一起讨论。毕竟，程序员最懂程序员，不是吗？😉
如何从零开始搭建FastAPI+SQLAlchemy 2.0+Alembic数据库实战项目？
摘要：本文以实战为导向，从零开始搭建 FastAPI + SQLAlchemy 2.0 + Alembic 的数据库层。通过生活化比喻和踩坑案例，讲透异步引擎配置、模型定义新写法、迁移脚本生成与审核等核心环节，