如何使用Alembic迁移已存在且有数据的SQLAlchemy Core定义表

实现步骤 #

1. 先调整现有代码结构

你当前把MetaData和Table定义写在类的__init__方法里，Alembic无法直接导入获取元数据，需要先把表结构定义抽离为全局可导入的对象，示例调整如下：

# 建议单独存为 db_models.py
from sqlalchemy import MetaData, Table, Column, Integer, String, DateTime, Boolean, func
from sqlalchemy import create_engine
from sqlalchemy.pool import NullPool

DB_ADDRESS = "你的PostgreSQL连接字符串"
engine = create_engine(DB_ADDRESS, poolclass=NullPool)
# 全局唯一的MetaData对象，所有表都绑定到这个实例
meta = MetaData()

# 表定义抽离到全局
employeetable = Table(
    "employeetable",
    meta,
    Column("id", Integer, primary_key=True),
    Column("employeeId", String),
    Column("name", String),
    Column("email", String),
    Column("place", String),
    Column("sector", String),
    Column("function", String),
    Column("phone", String),
    Column("status", String),
    Column("type", String),
    Column(
        "lastChange",
        DateTime(timezone=True),
        server_default=func.now(),
    ),
    Column("receptorId", String),
    Column("syncFlag", Boolean),
)

# 原封装的表操作类调整为引用全局的元数据和表对象
class EmployeeTable:
    def __init__(self):
        self.engine = engine
        self.meta = meta
        self.table = employeetable

    # 原有增删改查方法保留即可

2. 初始化并配置Alembic

• 安装Alembic：pip install alembic
• 项目根目录执行初始化命令：alembic init alembic
• 修改alembic.ini配置文件，将sqlalchemy.url的值改为你的PostgreSQL连接字符串，和上面的DB_ADDRESS保持一致
• 修改alembic/env.py文件：
  1. 开头加入项目根目录到Python路径，避免导入自定义模块失败：

  import sys, os
  sys.path.append(os.path.dirname(os.path.dirname(__file__)))
  

  2. 导入你定义的全局MetaData对象，替换默认的target_metadata：

  from db_models import meta
  target_metadata = meta
  

3. 生成基线迁移（适配已有数据的表）

因为你当前的表已经存在且有业务数据，需要先标记现有表结构为已同步，避免Alembic生成重复的建表语句：

1. 确保此时db_models.py里的表定义和数据库现有结构完全一致，先不要加新字段
2. 执行命令生成初始基线脚本：alembic revision --autogenerate -m "init baseline"
3. 执行命令标记基线脚本为已执行，不会实际修改数据库：alembic stamp head

4. 生成加字段的迁移脚本

1. 在db_models.py的employeetable定义里添加你需要的新同步状态字段，示例：

Column("xxxSyncStatus", Boolean, default=False, comment="某字段同步状态")

2. 执行命令自动生成迁移脚本：alembic revision --autogenerate -m "add xxx sync status column"
3. 手动检查生成的迁移脚本内容，确认和你预期的修改一致（Alembic自动生成的内容可能存在识别误差，需人工校验）
4. 执行迁移应用到数据库：alembic upgrade head

注意事项 #

• 生产环境执行迁移前必须备份数据库，避免迁移异常导致数据丢失
• 如果有多个表需要管理，所有表都绑定到同一个全局MetaData对象即可，Alembic会自动识别所有表的结构变更
• 不要修改已经应用过的历史迁移脚本，所有结构变更都应该通过新增迁移脚本实现

内容的提问来源于stack exchange，提问作者gsab