Alembic，一个神奇的 Python 库

Alembic 是一款轻量型的数据库迁移工具，它与 SQLAlchemy 一起共同为 Python 提供数据库管理与迁移支持，为关系型数据提供创建、管理、更改和调用的管理脚本，协助开发和运维人员在系统上线后对数据库进行在线管理。

安装

Alembic是同步工具，仅支持同步数据库驱动，比如MySQL 同步驱动pymysql。

Alembic是 SQLAlchemy 作者编写的数据库迁移工具，不用独立安装SQLAlchemy。

pip install alembic pymysql

创建迁移环境

使用 Alembic 之前需要先建立一个 Alembic 脚本环境。在工程目录下执行下面脚本

alembic init alembic

执行完命令后，会在工程目录下生成alembic目录和alembic.ini配置文件

├── alembic
│   ├── README
│   ├── env.py
│   ├── script.py.mako
│   └── versions
├── alembic.ini

alembic.ini - 配置文件
alembic/ - 迁移脚本目录
alembic/env.py - 迁移环境配置
alembic/script.py.mako - 迁移脚本模板
alembic/versions用于存放各个版本的迁移脚本。

配置设置

1、配置数据库连接

修改配置文件alembic.ini

[alembic]
# 方法1：直接在配置文件中设置（不推荐提交敏感信息）
sqlalchemy.url = driver://user:password@host:port/database

修改env.py 配置脚本

config = context.config
# # 导入你的 SQLAlchemy同步数据库连接
from database import SYNC_DATABASE_URL
config.set_main_option('sqlalchemy.url', SYNC_DATABASE_URL)

2、设置模型基类

打开脚本env.py

# 导入你的 SQLAlchemy Base
from database import  Base  # 根据你的项目结构调整

# 设置 target_metadata
target_metadata = Base.metadata

3、生成初始迁移脚本

alembic revision -m "init table migration"

在versions看到迁移脚本




    
alembic upgrade head

# 输出
# ...
# INFO  [alembic.runtime.migration] Running upgrade  -> 3c2156449d57, init table migration

查看数据库我们会看到alembic_version数据表。

通过上面步骤完成后，表示我们配置成功，并完成了初始脚本迁移

常用命令

# 基于模型更改自动生成迁移脚本
alembic revision --autogenerate -m "描述信息"

# 创建空白迁移脚本
alembic revision -m "描述信息"

# 升级到最新版本
alembic upgrade head

# 升级到特定版本
alembic upgrade ae1027a6acf

# 降级一个版本
alembic downgrade -1

# 降级到特定版本
alembic downgrade base  # 回退到初始状态

# 查看当前版本
alembic current

# 查看历史
alembic history

# 查看详细历史
alembic history --verbose

# 检查是否需要迁移
alembic check

实战操作

创建模型

# models.py
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.sql import func
from database import Base


class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, index=True, nullable=False)
    email = Column(String(100), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    full_name = Column(String(100))
    is_active = Column(Integer, default=1)
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, onupdate=func.now())

模型创建完成后，还有修改env.py文件

# from database Base 在导入基类后导入模型类
from models import User # 这行很重要，要不--autogenerate不起作用




    
alembic revision --autogenerate -m "create_user_table"

# 在verisions生成了迁移脚本
# Generating alembic\versions\a591b3f651be_create_user_table.py ...  done

数据库数据表变更

alembic upgrade head

# INFO  [alembic.runtime.migration] Running upgrade 3c2156449d57 -> a591b3f651be, create_user_table

修改字段、添加字段、删除字段

修改模型User的密码字段password

password = Column(String(255), nullable=False) #变更hashed_password为password
full_name = Column(String(100)) # 删除字段
deleted_at = Column(DateTime, default=None) # 新增软删除字段

执行自动生成脚本

alembic revision --autogenerate -m "edit_user_table"

上面我本意是把hashed_password 修改成passoword,自动生成迁移脚本直接变更为删除后重新添加了

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('users', sa.Column('password', sa.String(length=255), nullable=False))
    op.add_column('users', sa.Column('deleted_at', sa.DateTime(), nullable=True))
    op.drop_column('users', 'hashed_password')
    # ### end Alembic commands ###


def 


    
downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('users', sa.Column('hashed_password', mysql.VARCHAR(collation='utf8mb4_bin', length=255), nullable=False))
    op.drop_column('users', 'deleted_at')
    op.drop_column('users', 'password')
    # ### end Alembic commands ###

「如果有数据的时候一定要注意了」 ，这时候我们要通过alter_column手动修改字段名，字段属性等。

def upgrade() -> None:
    """Upgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column(
        'users',
        'hashed_password',
        new_column_name='password',
        existing_type=sa.String(255),
        existing_nullable=False
    )
    op.add_column('users', sa.Column('deleted_at', sa.DateTime(), nullable=True))
    op.drop_column('users', 'full_name')
    # ### end Alembic commands ###


def downgrade() -> None:
    """Downgrade schema."""
    # ### commands auto generated by Alembic - please adjust! ###
    op.alter_column(
        'users',
        'password',
        new_column_name='hashed_password',
        existing_type=sa.String(255),
        existing_nullable=False
    )
    op.drop_column('users', 'deleted_at')
    op.add_column('users', sa.Column('full_name', mysql.VARCHAR(collation='utf8mb4_bin', length=100), nullable=True))
    # ### end Alembic commands ###

常用操作参考

# 创建表



    
op.create_table()
op.create_index()
op.create_unique_constraint()
op.create_foreign_key()
op.create_primary_key()

# 修改表
op.add_column()
op.drop_column()
op.alter_column()
op.rename_table()

# 数据操作
op.execute("UPDATE table SET column = value")
op.bulk_insert(table, rows)

优劣势分析

维度	优势 ✅	劣势 ❌
集成度	与 SQLAlchemy 完美集成	仅适用于 SQLAlchemy 项目
易用性	学习曲线平缓	需要 Python/SQL 知识
功能	版本控制、回滚、分支管理	复杂数据迁移支持有限
兼容性	支持多种数据库后端	不同数据库特性支持不一
自动化	自动生成迁移脚本	自动生成有时不完美

对于新项目，并且使用SQLAlchemy 作为数据库ORM工具的，多人协作开发的时候我们可以使用Alembic做为迁移工具，它提供类似 Git 的版本控制系统，支持自动生成迁移脚本、灵活的升级降级操作和事务安全性。对于使用 SQLAlchemy 的 Python 项目，特别是在快速迭代开发阶段，Alembic 能显著提升数据库 schema 管理效率。