生成与执行迁移

一、工作流程

改 Model → 生成迁移脚本 → 检查脚本 → 执行迁移 → 数据库更新

# 1. 改 models.py
# 2. 生成迁移脚本
alembic revision --autogenerate -m "描述"
# 3. 检查 alembic/versions/ 下生成的脚本
# 4. 执行迁移
alembic upgrade head

二、autogenerate 能做的事

以下场景 Alembic 可以自动检测并生成迁移代码。

1. 创建新表

# models.py — 新增 User 模型
class User(Base):
    __tablename__ = "users"
    id    = Column(Integer, primary_key=True)
    name  = Column(String(50), nullable=False)
    email = Column(String(100), unique=True, nullable=False)

alembic revision --autogenerate -m "创建用户表"

生成的迁移脚本：

def upgrade() -> None:
    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.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
    )

def downgrade() -> None:
    op.drop_table('users')

2. 删除表

# models.py — 删掉 User 模型
# （直接把 class User 删掉就行）

alembic revision --autogenerate -m "删除用户表"

生成的迁移脚本：

def upgrade() -> None:
    op.drop_table('users')

def downgrade() -> None:
    op.create_table('users',
        sa.Column('id', sa.Integer(), nullable=False),
        # ... 所有字段都会还原回来
    )

3. 新增字段

# models.py — User 加一个 age 字段
class User(Base):
    __tablename__ = "users"
    id   = Column(Integer, primary_key=True)
    name = Column(String(50))
    age  = Column(Integer, default=0)        # 新增

alembic revision --autogenerate -m "用户表添加age字段"

生成的迁移脚本：

def upgrade() -> None:
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))

def downgrade() -> None:
    op.drop_column('users', 'age')

4. 删除字段

# models.py — User 删掉 email 字段
class User(Base):
    __tablename__ = "users"
    id   = Column(Integer, primary_key=True)
    name = Column(String(50))
    # email 删掉了

alembic revision --autogenerate -m "用户表删除email字段"

生成的迁移脚本：

def upgrade() -> None:
    op.drop_column('users', 'email')

def downgrade() -> None:
    op.add_column('users', sa.Column('email', sa.String(length=100), nullable=False))

5. 修改字段类型或长度

# models.py — name 从 String(50) 改成 String(100)
name = Column(String(100))   # 原来是 String(50)

alembic revision --autogenerate -m "用户表name字段长度改为100"

生成的迁移脚本：

def upgrade() -> None:
    op.alter_column('users', 'name', existing_type=sa.String(length=50),
                    type_=sa.String(length=100))

def downgrade() -> None:
    op.alter_column('users', 'name', existing_type=sa.String(length=100),
                    type_=sa.String(length=50))

6. 修改 nullable 约束

# models.py — name 从 nullable=False 改成 nullable=True
name = Column(String(50), nullable=True)   # 原来是 nullable=False

alembic revision --autogenerate -m "用户表name字段允许为空"

生成的迁移脚本：

def upgrade() -> None:
    op.alter_column('users', 'name', existing_type=sa.String(length=50), nullable=True)

def downgrade() -> None:
    op.alter_column('users', 'name', existing_type=sa.String(length=50), nullable=False)

7. 新增索引

# models.py — 给 name 加索引
name = Column(String(50), index=True)   # 新增 index=True

alembic revision --autogenerate -m "用户表name字段添加索引"

生成的迁移脚本：

def upgrade() -> None:
    op.create_index(op.f('ix_users_name'), 'users', ['name'], unique=False)

def downgrade() -> None:
    op.drop_index(op.f('ix_users_name'), table_name='users')

8. 新增唯一约束

# models.py — 给 name 加唯一约束
name = Column(String(50), unique=True)   # 新增 unique=True

alembic revision --autogenerate -m "用户表name字段添加唯一约束"

生成的迁移脚本：

def upgrade() -> None:
    op.create_unique_constraint(op.f('uq_users_name'), 'users', ['name'])

def downgrade() -> None:
    op.drop_constraint(op.f('uq_users_name'), 'users', type_='unique')

9. 新增外键

# models.py — Article 新增 author_id 外键
class Article(Base):
    __tablename__ = "articles"
    id        = Column(Integer, primary_key=True)
    title     = Column(String(200))
    author_id = Column(Integer, ForeignKey("users.id"))   # 新增

alembic revision --autogenerate -m "文章表添加author_id外键"

生成的迁移脚本：

def upgrade() -> None:
    op.add_column('articles', sa.Column('author_id', sa.Integer(), nullable=True))
    op.create_foreign_key(None, 'articles', 'users', ['author_id'], ['id'])

def downgrade() -> None:
    op.drop_constraint(None, 'articles', type_='foreignkey')
    op.drop_column('articles', 'author_id')

10. 小结：能做的事

操作	Model 改动	自动生成的命令
创建表	新增一个 Model 类	`op.create_table(...)`
删除表	删掉一个 Model 类	`op.drop_table(...)`
新增字段	加一个 Column	`op.add_column(...)`
删除字段	删一个 Column	`op.drop_column(...)`
改类型/长度	改 Column 的类型	`op.alter_column(..., type_=...)`
改 nullable	改 nullable 参数	`op.alter_column(..., nullable=...)`
新增索引	加 `index=True`	`op.create_index(...)`
删索引	去掉 `index=True`	`op.drop_index(...)`
新增唯一约束	加 `unique=True`	`op.create_unique_constraint(...)`
删唯一约束	去掉 `unique=True`	`op.drop_constraint(...)`
新增加键	加 `ForeignKey(...)`	`op.create_foreign_key(...)`
删外键	去掉 `ForeignKey`	`op.drop_constraint(...)`

三、autogenerate 不能做的事（及替代方案）

1. 重命名表

问题： Alembic 会把重命名当成「删旧表 + 建新表」，数据会丢失。

# models.py
# 原来：__tablename__ = "users"
# 改成：__tablename__ = "accounts"    ← Alembic 以为你删了 users、建了 accounts

替代方案： 手动编辑迁移脚本，用 op.rename_table()：

def upgrade() -> None:
    op.rename_table('users', 'accounts')

def downgrade() -> None:
    op.rename_table('accounts', 'users')

2. 重命名字段

问题： Alembic 会把重命名当成「删旧字段 + 建新字段」，数据会丢失。

# models.py
# 原来：name = Column(String(50))
# 改成：full_name = Column(String(50))    ← Alembic 以为你删了 name、建了 full_name

替代方案： 手动编辑迁移脚本，用 op.alter_column() 重命名：

def upgrade() -> None:
    op.alter_column('users', 'name', new_column_name='full_name')

def downgrade() -> None:
    op.alter_column('users', 'full_name', new_column_name='name')

不同数据库可能写法不同，MySQL 用：

def upgrade() -> None:
    op.alter_column('users', 'name', new_column_name='full_name',
                    existing_type=sa.String(length=50))

3. 数据迁移

问题： Alembic 只能改表结构，不能帮你迁移数据。

场景：把 name 字段拆成 first_name + last_name。

替代方案： 在迁移脚本中用 op.execute() 写 SQL：

def upgrade() -> None:
    # 1. 加新字段
    op.add_column('users', sa.Column('first_name', sa.String(50)))
    op.add_column('users', sa.Column('last_name', sa.String(50)))

    # 2. 迁移数据（用原生 SQL）
    op.execute("""
        UPDATE users
        SET first_name = SUBSTR(name, 1, 1),
            last_name = SUBSTR(name, 2)
    """)

    # 3. 删旧字段
    op.drop_column('users', 'name')


def downgrade() -> None:
    op.add_column('users', sa.Column('name', sa.String(50)))
    op.execute("""
        UPDATE users
        SET name = first_name || last_name
    """)
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')

4. 修改主键

问题： Alembic 不能自动检测主键的变更。

替代方案： 手动编写迁移脚本：

def upgrade() -> None:
    # 先删旧主键
    op.drop_constraint('users_pkey', 'users', type_='primary')
    # 再建新主键
    op.create_primary_key('users_pkey', 'users', ['id', 'name'])


def downgrade() -> None:
    op.drop_constraint('users_pkey', 'users', type_='primary')
    op.create_primary_key('users_pkey', 'users', ['id'])

5. 数据库特定操作

问题： 存储过程、触发器、视图等 Alembic 不认识。

替代方案： 直接写 SQL：

def upgrade() -> None:
    # 创建视图
    op.execute("""
        CREATE VIEW active_users AS
        SELECT * FROM users WHERE is_active = 1
    """)

    # 创建触发器（MySQL 示例）
    op.execute("""
        CREATE TRIGGER update_timestamp
        BEFORE UPDATE ON users
        FOR EACH ROW
        SET NEW.updated_at = NOW()
    """)


def downgrade() -> None:
    op.execute("DROP VIEW IF EXISTS active_users")
    op.execute("DROP TRIGGER IF EXISTS update_timestamp")

6. 跨库操作

问题： 一次迁移只能操作一个数据库。

替代方案： 分别配置两个 Alembic 实例（不同目录），各自管理各自的数据库。

7. 小结：不能做的事及替代方案

不能做的事	后果	替代方案
重命名表	被当成删旧建新，数据丢失	手动用 `op.rename_table()`
重命名字段	被当成删旧建新，数据丢失	手动用 `op.alter_column(new_column_name=...)`
数据迁移	只改结构不改数据	在迁移脚本中用 `op.execute()` 写 SQL
修改主键	不能自动检测	手动 `op.drop_constraint` + `op.create_primary_key`
存储过程/触发器/视图	不认识	用 `op.execute()` 写原生 SQL
跨库操作	一次只能一个库	分别配置 Alembic 实例

四、手动编写迁移脚本

有些场景需要手写迁移脚本：autogenerate 搞不定的（重命名、数据迁移），或者需要插入初始数据。

4.1 迁移脚本长什么样

先看懂自动生成的文件结构，才知道怎么改。

执行 alembic revision --autogenerate -m "创建用户表" 后，会在 alembic/versions/ 下生成一个 .py 文件：

alembic/
└── versions/
    └── abc123_创建用户表.py     ← 这就是迁移脚本

打开文件，内容是这样的：

"""创建用户表

Revision ID: abc123
Revises: 
Create Date: 2024-01-15 10:30:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa

# ---------- 版本标识（不用管） ----------
revision: str = 'abc123'
down_revision: Union[str, None] = None
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None


# ---------- 升级：执行 alembic upgrade 时运行 ----------
def upgrade() -> None:
    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.PrimaryKeyConstraint('id'),
        sa.UniqueConstraint('email'),
    )


# ---------- 降级：执行 alembic downgrade 时运行 ----------
def downgrade() -> None:
    op.drop_table('users')

你需要关注的只有两个函数：

函数	什么时候执行	作用
`upgrade()`	`alembic upgrade`	向前执行（升级）
`downgrade()`	`alembic downgrade`	向后回滚（降级）

其他内容（revision、down_revision 等）是 Alembic 用来管理版本链的，不需要动。

4.2 手动编辑迁移脚本的步骤

场景： 你想给 users 表加一个 age 字段，但同时要把已有数据的 age 设为 0。

第一步：正常生成迁移脚本

# 先在 models.py 里加上 age 字段
alembic revision --autogenerate -m "用户表添加age字段"

第二步：找到生成的文件

ls alembic/versions/
# 输出类似：abc123_用户表添加age字段.py

第三步：用编辑器打开文件

# 自动生成的内容（只有加字段，没有数据迁移）
def upgrade() -> None:
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))

def downgrade() -> None:
    op.drop_column('users', 'age')

第四步：手动补充内容

在 upgrade() 里加上数据迁移的 SQL：

def upgrade() -> None:
    # 自动生成的：加字段
    op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))

    # ↓↓↓ 手动补充的部分 ↓↓↓
    # 给已有数据设置默认值
    op.execute("UPDATE users SET age = 0 WHERE age IS NULL")
    # 把字段改成非空
    op.alter_column('users', 'age', nullable=False)

def downgrade() -> None:
    op.drop_column('users', 'age')

第五步：保存文件，执行迁移

alembic upgrade head

4.3 常用的 op 命令

手动编辑时最常用的命令：

表操作：

# 创建表
op.create_table('users',
    sa.Column('id', sa.Integer(), primary_key=True),
    sa.Column('name', sa.String(50), nullable=False),
)

# 删除表
op.drop_table('users')

# 重命名表
op.rename_table('users', 'accounts')

字段操作：

# 加字段
op.add_column('users', sa.Column('age', sa.Integer(), nullable=True))

# 删字段
op.drop_column('users', 'age')

# 重命名字段
op.alter_column('users', 'name', new_column_name='full_name')

# 改类型
op.alter_column('users', 'name',
    existing_type=sa.String(50),
    type_=sa.String(100))

# 改是否允许为空
op.alter_column('users', 'age', nullable=False)

约束操作：

# 加外键
op.create_foreign_key('fk_article_user', 'articles', 'users', ['author_id'], ['id'])

# 删外键
op.drop_constraint('fk_article_user', 'articles', type_='foreignkey')

# 加唯一约束
op.create_unique_constraint('uq_users_email', 'users', ['email'])

# 删唯一约束
op.drop_constraint('uq_users_email', 'users', type_='unique')

# 加索引
op.create_index('ix_users_name', 'users', ['name'])

# 删索引
op.drop_index('ix_users_name', table_name='users')

执行原生 SQL：

# 插入数据
op.execute("INSERT INTO users (name, email) VALUES ('admin', 'admin@example.com')")

# 更新数据
op.execute("UPDATE users SET age = 0 WHERE age IS NULL")

# 删除数据
op.execute("DELETE FROM users WHERE is_deleted = 1")

# 任意 SQL
op.execute("CREATE VIEW active_users AS SELECT * FROM users WHERE is_active = 1")

4.4 完整示例：重命名字段

场景： 把 users 表的 name 字段改名为 full_name。

第一步：改 models.py

# 原来
name = Column(String(50))

# 改成
full_name = Column(String(50))

第二步：生成迁移脚本

alembic revision --autogenerate -m "重命名name为full_name"

第三步：查看自动生成的内容（有问题）

def upgrade() -> None:
    # Alembic 以为你删了 name、建了 full_name
    op.drop_column('users', 'name')          # ← 数据会丢！
    op.add_column('users', sa.Column('full_name', sa.String(50)))

def downgrade() -> None:
    op.drop_column('users', 'full_name')
    op.add_column('users', sa.Column('name', sa.String(50)))

第四步：手动改成正确的写法

def upgrade() -> None:
    # 重命名，保留数据
    op.alter_column('users', 'name', new_column_name='full_name')

def downgrade() -> None:
    op.alter_column('users', 'full_name', new_column_name='name')

第五步：执行

alembic upgrade head

4.5 完整示例：数据迁移

场景： 把 users 表的 name 拆成 first_name 和 last_name。

第一步：改 models.py

# 删掉
# name = Column(String(50))

# 新增
first_name = Column(String(50))
last_name  = Column(String(50))

第二步：生成迁移脚本

alembic revision --autogenerate -m "拆分name为first_name和last_name"

第三步：查看自动生成的内容（不完整）

def upgrade() -> None:
    op.drop_column('users', 'name')
    op.add_column('users', sa.Column('first_name', sa.String(50)))
    op.add_column('users', sa.Column('last_name', sa.String(50)))
    # ← 缺少数据迁移！旧数据会丢失

第四步：手动补充数据迁移

def upgrade() -> None:
    # 1. 先加新字段
    op.add_column('users', sa.Column('first_name', sa.String(50)))
    op.add_column('users', sa.Column('last_name', sa.String(50)))

    # 2. 迁移数据：把 name 拆到 first_name 和 last_name
    op.execute("""
        UPDATE users
        SET first_name = SUBSTR(name, 1, 1),
            last_name = SUBSTR(name, 2)
    """)

    # 3. 最后删旧字段
    op.drop_column('users', 'name')


def downgrade() -> None:
    # 回滚：反过来操作
    op.add_column('users', sa.Column('name', sa.String(50)))
    op.execute("""
        UPDATE users
        SET name = first_name || last_name
    """)
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')

第五步：执行

alembic upgrade head

4.6 完整示例：插入初始数据

场景： 表建好了，想插入一些默认数据。

第一步：创建空迁移脚本（不用 --autogenerate）

alembic revision -m "插入初始数据"

第二步：打开文件，手动写内容

def upgrade() -> None:
    # 插入默认角色
    op.execute("""
        INSERT INTO roles (name, description) VALUES
        ('admin', '管理员'),
        ('user', '普通用户'),
        ('guest', '访客')
    """)

    # 插入默认管理员
    op.execute("""
        INSERT INTO users (name, email, is_active)
        VALUES ('admin', 'admin@example.com', 1)
    """)


def downgrade() -> None:
    # 回滚：删掉插入的数据
    op.execute("DELETE FROM users WHERE name = 'admin'")
    op.execute("DELETE FROM roles WHERE name IN ('admin', 'user', 'guest')")

第三步：执行

alembic upgrade head

五、执行迁移

5.1 升级

# 升级到最新版本
alembic upgrade head

# 升级一个版本
alembic upgrade +1

# 升级到指定版本
alembic upgrade abc123

5.2 降级（回滚）

# 降级一个版本
alembic downgrade -1

# 降级到最初状态（清空所有表结构变更）
alembic downgrade base

# 降级到指定版本
alembic downgrade abc123

5.3 查看状态

# 查看当前数据库在哪个版本
alembic current

# 查看所有迁移历史
alembic history --verbose

# 查看有哪些待执行的迁移
alembic history

六、动态配置数据库地址

不想在 alembic.ini 里写死数据库地址，在 env.py 里动态设置：

from database import SQLALCHEMY_DATABASE_URL

def run_migrations_online() -> None:
    # 覆盖 alembic.ini 中的配置
    config.set_main_option("sqlalchemy.url", SQLALCHEMY_DATABASE_URL)

    connectable = engine_from_config(
        config.get_section(config.config_ini_section, {}),
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    # ...

这样 Alembic 和 FastAPI 共用同一个数据库地址，改一处就行。

七、常见问题

1. 迁移脚本生成了但数据库没变化

# 查看当前版本
alembic current

如果显示的版本不是最新的，说明之前的迁移还没执行，先 alembic upgrade head。

2. 想重新生成迁移脚本

# 方式一：还没执行过，直接删文件重新生成
rm alembic/versions/xxx_*.py
alembic revision --autogenerate -m "重新生成"

# 方式二：已经执行了，先回滚再删文件
alembic downgrade -1
rm alembic/versions/xxx_*.py
alembic revision --autogenerate -m "重新生成"
alembic upgrade head

3. 报错 "Can't locate revision"

数据库里记录的版本号在 versions/ 目录里找不到。通常是因为删了迁移脚本但数据库里还有记录。

# 解决方式：删掉数据库重建
rm test.db
alembic upgrade head

4. 多人协作时版本冲突

两个人各自生成了迁移脚本，版本链分叉了。

# 查看历史，找到分叉点
alembic history --verbose

# 合并两个分支
alembic merge -m "合并两个迁移" rev1 rev2

#生成与执行迁移

#一、工作流程

#二、autogenerate 能做的事

#1. 创建新表

#2. 删除表

#3. 新增字段

#4. 删除字段

#5. 修改字段类型或长度

#6. 修改 nullable 约束

#7. 新增索引

#8. 新增唯一约束

#9. 新增外键

#10. 小结：能做的事

#三、autogenerate 不能做的事（及替代方案）

#1. 重命名表

#2. 重命名字段

#3. 数据迁移

#4. 修改主键

#5. 数据库特定操作

#6. 跨库操作

#7. 小结：不能做的事及替代方案

#四、手动编写迁移脚本

#4.1 迁移脚本长什么样

#4.2 手动编辑迁移脚本的步骤

#4.3 常用的 op 命令

#4.4 完整示例：重命名字段

#4.5 完整示例：数据迁移

#4.6 完整示例：插入初始数据

#五、执行迁移

#5.1 升级

#5.2 降级（回滚）

#5.3 查看状态

#六、动态配置数据库地址

#七、常见问题

#1. 迁移脚本生成了但数据库没变化

#2. 想重新生成迁移脚本

#3. 报错 "Can't locate revision"

#4. 多人协作时版本冲突

生成与执行迁移

一、工作流程

二、autogenerate 能做的事

1. 创建新表

2. 删除表

3. 新增字段

4. 删除字段

5. 修改字段类型或长度

6. 修改 nullable 约束

7. 新增索引

8. 新增唯一约束

9. 新增外键

10. 小结：能做的事

三、autogenerate 不能做的事（及替代方案）

1. 重命名表

2. 重命名字段

3. 数据迁移

4. 修改主键

5. 数据库特定操作

6. 跨库操作

7. 小结：不能做的事及替代方案

四、手动编写迁移脚本

4.1 迁移脚本长什么样

4.2 手动编辑迁移脚本的步骤

4.3 常用的 op 命令

4.4 完整示例：重命名字段

4.5 完整示例：数据迁移

4.6 完整示例：插入初始数据

五、执行迁移

5.1 升级

5.2 降级（回滚）

5.3 查看状态

六、动态配置数据库地址

七、常见问题

1. 迁移脚本生成了但数据库没变化

2. 想重新生成迁移脚本

3. 报错 "Can't locate revision"

4. 多人协作时版本冲突