PorterYuan · 个人学习博客

Alembic 数据库迁移：异步引擎配置 + upgrade/downgrade + 自动导入 Model

9分钟阅读

Alembic 数据库迁移：异步引擎配置 + upgrade/downgrade + 自动导入 Model

一句话：Alembic是SQLAlchemy的数据库迁移工具，可以管理数据库结构的变更。在FastAPI项目中，Alembic常用于管理数据库schema的版本控制。

1. Alembic 基础

1.1 什么是Alembic？


graph LR

    A[Alembic] --> B[数据库迁移工具]

    A --> C[版本控制]

    A --> D[自动生成迁移脚本]

    A --> E[支持升级/降级]

    style A fill:#e1f5fe

Alembic：SQLAlchemy的数据库迁移工具，用于管理数据库结构的变更。

1.2 安装和初始化

 
# 安装
 
pip install alembic
 
# 初始化
 
alembic init alembic

1.3 配置文件

 
# alembic.ini
 
[alembic]
 
script_location = alembic
 
sqlalchemy.url = postgresql://user:pass@localhost/db
 
# sqlalchemy.url = %(DATABASE_URL)s

2. 异步引擎配置

2.1 异步配置

 
# alembic/env.py
 
import asyncio
 
from logging.config import fileConfig
 
from sqlalchemy import pool
 
from sqlalchemy.engine import Connection
 
from sqlalchemy.ext.asyncio import async_engine_from_config
 
from alembic import context
 
# this is the Alembic Config object
 
config = context.config
 
# This line sets up loggers basically.
 
if config.config_file_name is not None:
 
    fileConfig(config.config_file_name)
 
# for 'autogenerate' support
 
from app.models import Base
 
target_metadata = Base.metadata
 
def run_migrations_offline() -> None:
 
    """Run migrations in 'offline' mode."""
 
    url = config.get_main_option("sqlalchemy.url")
 
    context.configure(
 
        url=url,
 
        target_metadata=target_metadata,
 
        literal_binds=True,
 
        dialect_opts={"paramstyle": "named"},
 
    )
 
    with context.begin_transaction():
 
        context.run_migrations()
 
def do_run_migrations(connection: Connection) -> None:
 
    context.configure(connection=connection, target_metadata=target_metadata)
 
    with context.begin_transaction():
 
        context.run_migrations()
 
async def run_async_migrations() -> None:
 
    """Run migrations in 'online' mode with async engine."""
 
    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)
 
    await connectable.dispose()
 
def run_migrations_online() -> None:
 
    """Run migrations in 'online' mode."""
 
    asyncio.run(run_async_migrations())
 
if context.is_offline_mode():
 
    run_migrations_offline()
 
else:
 
    run_migrations_online()

3. 生成迁移脚本

3.1 自动生成

 
# 自动生成迁移脚本
 
alembic revision --autogenerate -m "添加用户表"
 
# 生成空迁移脚本
 
alembic revision -m "自定义迁移"

3.2 迁移脚本示例

 
# alembic/versions/xxx_添加用户表.py
 
"""添加用户表
 
Revision ID: xxx
 
Revises:
 
Create Date: 2026-07-21 12:00:00.000000
 
"""
 
from alembic import op
 
import sqlalchemy as sa
 
# revision identifiers, used by Alembic.
 
revision = 'xxx'
 
down_revision = None
 
branch_labels = None
 
depends_on = None
 
def upgrade() -> None:
 
    # ### commands auto generated by Alembic - please adjust! ###
 
    op.create_table('users',
 
    sa.Column('id', sa.Integer(), nullable=False),
 
    sa.Column('name', sa.String(length=50), nullable=False),
 
    sa.Column('email', sa.String(length=100), nullable=False),
 
    sa.Column('created_at', sa.DateTime(), nullable=True),
 
    sa.PrimaryKeyConstraint('id')
 
    )
 
    op.create_index(op.f('ix_users_email'), 'users', ['email'], unique=True)
 
    # ### end Alembic commands ###
 
def downgrade() -> None:
 
    # ### commands auto generated by Alembic - please adjust! ###
 
    op.drop_index(op.f('ix_users_email'), table_name='users')
 
    op.drop_table('users')
 
    # ### end Alembic commands ###

4. 执行迁移

4.1 升级数据库

 
# 升级到最新版本
 
alembic upgrade head
 
# 升级一个版本
 
alembic upgrade +1
 
# 升级到指定版本
 
alembic upgrade xxx

4.2 降级数据库

 
# 降级一个版本
 
alembic downgrade -1
 
# 降级到指定版本
 
alembic downgrade xxx
 
# 降级到初始版本
 
alembic downgrade base

4.3 查看版本历史

 
# 查看当前版本
 
alembic current
 
# 查看版本历史
 
alembic history
 
# 查看指定版本的迁移脚本
 
alembic history -v

5. 自动导入 Model

5.1 自动导入配置

 
# alembic/env.py
 
import os
 
import sys
 
from pathlib import Path
 
# 将项目根目录添加到Python路径
 
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
# 导入所有模型
 
from app.models.user import User
 
from app.models.post import Post
 
# 或者使用自动导入
 
from app.models import Base

5.2 模型定义示例

 
# app/models/user.py
 
from sqlalchemy import Column, Integer, String, DateTime
 
from sqlalchemy.orm import relationship
 
from datetime import datetime
 
from app.database import Base
 
class User(Base):
 
    __tablename__ = "users"
 
    id = Column(Integer, primary_key=True, index=True)
 
    name = Column(String(50), nullable=False)
 
    email = Column(String(100), unique=True, index=True, nullable=False)
 
    created_at = Column(DateTime, default=datetime.utcnow)
 
    # 关系
 
    posts = relationship("Post", back_populates="author")

6. 高级用法

6.1 数据迁移

 
# 迁移脚本中的数据迁移
 
def upgrade() -> None:
 
    # 创建新表
 
    op.create_table('new_users',
 
        sa.Column('id', sa.Integer(), nullable=False),
 
        sa.Column('full_name', sa.String(length=100), nullable=False),
 
        sa.PrimaryKeyConstraint('id')
 
    )
 
    # 迁移数据
 
    op.execute("""
 
        INSERT INTO new_users (id, full_name)
 
        SELECT id, name || ' ' || last_name
 
        FROM users
 
    """)
 
    # 删除旧表
 
    op.drop_table('users')
 
    # 重命名新表
 
    op.rename_table('new_users', 'users')

6.2 条件迁移

 
def upgrade() -> None:
 
    # 根据数据库类型执行不同操作
 
    dialect = op.get_context().dialect.name
 
    if dialect == 'postgresql':
 
        # PostgreSQL特定操作
 
        op.execute("CREATE EXTENSION IF NOT EXISTS pg_trgm")
 
    elif dialect == 'mysql':
 
        # MySQL特定操作
 
        pass

6.3 迁移测试

 
# 测试迁移脚本
 
def test_migration():
 
    # 升级到最新版本
 
    op.execute("alembic upgrade head")
 
    # 验证表结构
 
    result = op.get_bind().execute("SELECT * FROM users")
 
    assert result.fetchone() is not None
 
    # 降级
 
    op.execute("alembic downgrade -1")

7. 实际案例

7.1 FastAPI项目集成

 
# app/database.py
 
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
 
from sqlalchemy.orm import sessionmaker
 
DATABASE_URL = "postgresql+asyncpg://user:pass@localhost/db"
 
engine = create_async_engine(DATABASE_URL)
 
async_session = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
 
# sqlalchemy.url = postgresql+asyncpg://user:pass@localhost/db

8. 常见坑点

1. 迁移脚本冲突

 
# 解决：及时拉取最新迁移脚本，解决冲突
 
git pull origin main
 
alembic merge -m "合并冲突"

2. 数据丢失

 
# 解决：在迁移脚本中备份数据
 
def downgrade() -> None:
 
    # 备份数据
 
    op.execute("CREATE TABLE users_backup AS SELECT * FROM users")
 
    # 执行降级
 
    op.drop_table('users')

3. 异步引擎配置错误

 
# sqlalchemy.url = postgresql://user:pass@localhost/db

核心要点

 
# 初始化
 
alembic init alembic
 
# 生成迁移
 
alembic revision --autogenerate -m "描述"
 
# 执行迁移
 
alembic upgrade head  # 升级到最新
 
alembic downgrade -1  # 降级一个版本
 
# 查看版本
 
alembic current  # 当前版本
 
alembic history  # 版本历史

速记卡（面试闪卡）

Q1：一句话讲清「Alembic 数据库迁移：异步引擎配置 + upgrade/downgrade + 自动导入 Model」到底是什么？

A：Alembic 是 SQLAlchemy 的数据库迁移工具，像给库结构写带版本号的变更日记，升级降级可追溯。

Q2：Alembic 是什么（migration） —— 怎么理解？

A：类比：把数据库结构想象成乐高说明书：每次改动都生成一页带编号的”修订页”（revision），队友照着翻到第几版就能复现同样的积木造型。Alembic 就是管这批修订页的版本控制系统。（Versioned schema）

Q3：异步引擎怎么配（async engine） —— 怎么理解？

A：类比：普通连接像打电话要等人接，异步引擎（async engine）像发微信——await 一下就去干别的，连接空档不浪费。env.py 里用 async_engine_from_config 配好 asyncpg 驱动，迁移时连接池不卡主线程。（Non-blocking IO）

Q4：自动生成迁移脚本（autogenerate） —— 怎么理解？

A：类比：你改了 Model 类，Alembic 像勤快秘书自动对比当前库和代码差异，敲一行 alembic revision --autogenerate 就起草好 upgrade/downgrade 两个方向变更稿，你只需校对签字。（Diff & draft）

Q5：自动导入 Model（import Base） —— 怎么理解？

A：类比：Model 散落各处，Alembic 翻译时得先认识它们。把项目根塞进 sys.path 再 import 所有 Model（或挂到 Base.metadata），就像给翻译官发全公司花名册，生成 SQL 才不会”查无此人”漏表。（Register metadata）

Q6：核心速记主线有哪些？

  • Alembic 是 SQLAlchemy 的迁移/版本控制工具（migration）

  • 异步引擎用 async_engine_from_config + asyncpg，不堵主线程

  • revision —autogenerate 自动比对 Model 与库差异起草脚本

  • upgrade head 前进 / downgrade -1 回滚，current/history 看版本

  • 自动导入 Model 靠 sys.path + import Base.metadata，避免漏表

口诀

A：Alembic 管版本，迁移 revision；

升级前进 head，回滚 downgrade；

异步配 asyncpg，autogenerate 起草；

Model 先导入，改表错不了。

相关链接

关系图谱

反向链接

近期更新