Park Lam的每日分享

Python数据库迁移利器：Alembic全面使用教程

作者：

parklam

在

一、Alembic简介

Alembic是SQLAlchemy作者开发的数据库迁移工具，用于管理数据库模式变更。它能追踪模型变化，生成迁移脚本，支持版本控制和回滚操作。工作原理基于SQLAlchemy的元数据反射，通过对比模型与数据库结构生成差异脚本。

优点：与SQLAlchemy无缝集成，支持多种数据库，迁移脚本可手动编辑。缺点：初期配置稍复杂，对新手不够友好。Alembic采用MIT许可证，允许自由使用和修改。

二、Alembic安装与初始化

2.1 安装Alembic

使用pip可以轻松安装Alembic：

pip install alembic

安装完成后，可以通过以下命令验证安装是否成功：

alembic --version

如果安装成功，会显示当前Alembic的版本信息。

2.2 初始化Alembic环境

在你的项目目录中，执行以下命令初始化Alembic环境：

alembic init alembic

这个命令会在当前目录下创建一个名为alembic的文件夹和一个alembic.ini配置文件。初始化成功后，你的项目结构会类似这样：

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

其中，alembic.ini是主配置文件，alembic文件夹包含迁移脚本和环境配置。

2.3 配置数据库连接

编辑alembic.ini文件，找到sqlalchemy.url配置项，设置你的数据库连接字符串。例如，对于SQLite数据库：

sqlalchemy.url = sqlite:///mydatabase.db

对于PostgreSQL数据库：

sqlalchemy.url = postgresql://user:password@localhost/mydatabase

对于MySQL数据库：

sqlalchemy.url = mysql+pymysql://user:password@localhost/mydatabase

你也可以在alembic/env.py文件中通过代码配置数据库连接，这在需要动态配置的情况下非常有用：

# 在alembic/env.py中
from myapp import create_app
from myapp.models import Base

app = create_app()
target_metadata = Base.metadata

def run_migrations_online():
    connectable = app.engine  # 从应用中获取引擎

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

三、Alembic基本使用方法

3.1 创建迁移脚本

Alembic提供了两种创建迁移脚本的方式：自动生成和手动创建。

3.1.1 自动生成迁移脚本

当你已经定义了SQLAlchemy模型，并希望根据模型生成迁移脚本时，可以使用以下命令：

alembic revision --autogenerate -m "描述迁移的信息"

例如，如果你创建了一个用户模型，可以运行：

alembic revision --autogenerate -m "add user table"

这个命令会在alembic/versions目录下生成一个新的迁移脚本文件，文件名格式为{版本号}_{描述}.py。

自动生成的脚本会包含两个主要函数：upgrade()和downgrade()。upgrade()函数用于应用迁移，downgrade()函数用于回滚迁移。

3.1.2 手动创建迁移脚本

如果你需要手动编写迁移脚本，可以使用以下命令创建一个空的迁移脚本：

alembic revision -m "描述迁移的信息"

然后编辑生成的脚本文件，手动编写upgrade()和downgrade()函数中的逻辑。

例如，手动创建一个添加用户表的迁移脚本：

"""add user table

Revision ID: 1234567890ab
Revises: 
Create Date: 2023-07-15 10:00:00.000000

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic.
revision = '1234567890ab'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('username', sa.String(length=50), nullable=False, unique=True),
        sa.Column('email', sa.String(length=100), nullable=False, unique=True),
        sa.Column('password_hash', sa.String(length=255), nullable=False),
        sa.Column('created_at', sa.DateTime(), default=sa.func.now())
    )


def downgrade():
    op.drop_table('users')

3.2 应用迁移

创建迁移脚本后，可以使用以下命令将迁移应用到数据库：

alembic upgrade head

这个命令会将所有未应用的迁移脚本按顺序执行，将数据库更新到最新版本。

你也可以指定迁移到特定版本：

alembic upgrade 1234567890ab

或者相对于当前版本升级一定数量的迁移：

alembic upgrade +2

3.3 回滚迁移

如果需要回滚迁移，可以使用downgrade命令。回滚到上一个版本：

alembic downgrade -1

回滚到特定版本：

alembic downgrade 0987654321fe

回滚到最初始的版本：

alembic downgrade base

3.4 查看迁移历史

可以使用以下命令查看所有迁移版本的历史记录：

alembic history

加上-v参数可以查看更详细的信息：

alembic history -v

查看当前数据库的版本：

alembic current

四、Alembic高级用法

4.1 批量操作

当需要对多个表进行操作时，可以使用Alembic的批量操作API，它提供了更灵活的表结构修改方式，并且在不同数据库之间有更好的兼容性。

例如，批量添加列到多个表：

from alembic import op
import sqlalchemy as sa
from alembic.batch_alter_table import BatchOperations, batch_alter_table

def upgrade():
    # 定义要添加的列
    new_columns = [
        sa.Column('updated_at', sa.DateTime(), default=sa.func.now(), onupdate=sa.func.now())
    ]

    # 要添加列的表列表
    tables = ['users', 'posts', 'comments']

    for table in tables:
        with batch_alter_table(table) as batch_op:
            for column in new_columns:
                batch_op.add_column(column)

def downgrade():
    # 要删除的列
    columns_to_drop = ['updated_at']

    # 要操作的表列表
    tables = ['users', 'posts', 'comments']

    for table in tables:
        with batch_alter_table(table) as batch_op:
            for column in columns_to_drop:
                batch_op.drop_column(column)

4.2 数据迁移

除了结构迁移，Alembic也可以用于数据迁移。例如，在修改表结构前先迁移数据：

from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker

# 定义临时模型，用于数据迁移
class OldUser(sa.ext.declarative.Base):
    __tablename__ = 'users'
    id = sa.Column(sa.Integer, primary_key=True)
    full_name = sa.Column(sa.String(100))

class NewUser(sa.ext.declarative.Base):
    __tablename__ = 'users'
    id = sa.Column(sa.Integer, primary_key=True)
    first_name = sa.Column(sa.String(50))
    last_name = sa.Column(sa.String(50))

def upgrade():
    # 先添加新列
    op.add_column('users', sa.Column('first_name', sa.String(50)))
    op.add_column('users', sa.Column('last_name', sa.String(50)))

    # 创建会话
    Session = sessionmaker()
    bind = op.get_bind()
    session = Session(bind=bind)

    # 迁移数据：将full_name拆分为first_name和last_name
    for user in session.query(OldUser):
        if user.full_name:
            name_parts = user.full_name.split(' ', 1)
            user.first_name = name_parts[0]
            user.last_name = name_parts[1] if len(name_parts) > 1 else ''

    session.commit()

    # 删除旧列
    op.drop_column('users', 'full_name')

def downgrade():
    # 添加回旧列
    op.add_column('users', sa.Column('full_name', sa.String(100)))

    # 创建会话
    Session = sessionmaker()
    bind = op.get_bind()
    session = Session(bind=bind)

    # 恢复数据：将first_name和last_name合并为full_name
    for user in session.query(NewUser):
        user.full_name = f"{user.first_name} {user.last_name}".strip()

    session.commit()

    # 删除新列
    op.drop_column('users', 'first_name')
    op.drop_column('users', 'last_name')

4.3 事务管理

Alembic默认会在事务中执行迁移操作，但你也可以根据需要手动管理事务。

from alembic import op
import sqlalchemy as sa

def upgrade():
    # 禁用自动事务管理
    connection = op.get_bind()
    transaction = connection.begin()

    try:
        # 执行迁移操作
        op.create_table('categories',
            sa.Column('id', sa.Integer(), primary_key=True),
            sa.Column('name', sa.String(50), nullable=False)
        )

        # 手动提交事务
        transaction.commit()
    except Exception as e:
        # 发生错误时回滚
        transaction.rollback()
        raise e

def downgrade():
    connection = op.get_bind()
    transaction = connection.begin()

    try:
        op.drop_table('categories')
        transaction.commit()
    except Exception as e:
        transaction.rollback()
        raise e

4.4 环境变量配置

在实际项目中，数据库连接信息通常不会硬编码在配置文件中，而是通过环境变量获取。可以修改alembic/env.py文件来支持环境变量：

# 在alembic/env.py中
import os
from dotenv import load_dotenv  # 需要安装python-dotenv包
from sqlalchemy import create_engine

# 加载环境变量
load_dotenv()

# 从环境变量获取数据库连接信息
DB_USER = os.getenv('DB_USER')
DB_PASSWORD = os.getenv('DB_PASSWORD')
DB_HOST = os.getenv('DB_HOST', 'localhost')
DB_PORT = os.getenv('DB_PORT', '5432')
DB_NAME = os.getenv('DB_NAME')

# 构建数据库连接字符串
SQLALCHEMY_DATABASE_URL = f"postgresql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"

# 配置目标元数据
from myapp.models import Base
target_metadata = Base.metadata

def run_migrations_online():
    connectable = create_engine(SQLALCHEMY_DATABASE_URL)

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata
        )

        with context.begin_transaction():
            context.run_migrations()

然后创建一个.env文件存储数据库连接信息：

DB_USER=myuser
DB_PASSWORD=mypassword
DB_HOST=localhost
DB_PORT=5432
DB_NAME=mydatabase

这样就可以避免在代码中硬编码敏感信息。

五、实际项目案例

假设我们正在开发一个博客系统，需要使用Alembic管理数据库迁移。以下是整个过程的示例：

5.1 项目结构

blog_project/
├── alembic/
├── alembic.ini
├── .env
├── models.py
└── app.py

5.2 定义数据模型

首先，在models.py中定义我们的数据库模型：

from sqlalchemy import Column, Integer, String, Text, DateTime, ForeignKey
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship
from datetime import datetime

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String(50), unique=True, nullable=False)
    email = Column(String(100), unique=True, nullable=False)
    password_hash = Column(String(255), nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)

    # 关系
    posts = relationship('Post', back_populates='author')

class Post(Base):
    __tablename__ = 'posts'

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    author_id = Column(Integer, ForeignKey('users.id'))

    # 关系
    author = relationship('User', back_populates='posts')
    comments = relationship('Comment', back_populates='post')

class Comment(Base):
    __tablename__ = 'comments'

    id = Column(Integer, primary_key=True)
    content = Column(Text, nullable=False)
    created_at = Column(DateTime, default=datetime.utcnow)
    post_id = Column(Integer, ForeignKey('posts.id'))
    author_id = Column(Integer, ForeignKey('users.id'))

    # 关系
    post = relationship('Post', back_populates='comments')
    author = relationship('User')

5.3 初始化并配置Alembic

初始化Alembic环境：

alembic init alembic

编辑alembic.ini文件，配置数据库连接（或者使用前面介绍的环境变量方式）：

sqlalchemy.url = postgresql://myuser:mypassword@localhost/blogdb

修改alembic/env.py文件，指定目标元数据：

# 在alembic/env.py中
from models import Base
target_metadata = Base.metadata

5.4 创建初始迁移

生成初始迁移脚本：

alembic revision --autogenerate -m "initial schema"

这会生成一个包含创建所有表的迁移脚本。检查生成的脚本无误后，应用迁移：

alembic upgrade head

5.5 模型变更与迁移

随着项目发展，我们需要对模型进行修改。例如，我们想给用户添加一个bio字段：

# 在User模型中添加
bio = Column(Text, nullable=True)

生成新的迁移脚本：

alembic revision --autogenerate -m "add user bio"

检查生成的脚本，确认它包含添加bio列的操作，然后应用迁移：

alembic upgrade head

5.6 数据迁移案例

假设我们需要将Post表的title字段长度从200增加到300，并且需要对现有数据进行处理（如果标题过长则截断）：

# 首先修改模型
title = Column(String(300), nullable=False)  # 从200改为300

生成迁移脚本：

alembic revision --autogenerate -m "increase post title length"

然后编辑生成的迁移脚本，添加数据处理逻辑：

"""increase post title length

Revision ID: 5f3a7b9d1c2e
Revises: previous_revision_id
Create Date: 2023-07-16 14:30:00.000000

"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.orm import sessionmaker

# 定义临时模型用于数据处理
class Post(sa.ext.declarative.Base):
    __tablename__ = 'posts'
    id = sa.Column(sa.Integer, primary_key=True)
    title = sa.Column(sa.String(200))  # 原始长度

def upgrade():
    # 1. 先添加一个临时列
    op.add_column('posts', sa.Column('new_title', sa.String(300)))

    # 2. 截断过长的标题并迁移到临时列
    bind = op.get_bind()
    Session = sessionmaker(bind=bind)
    session = Session()

    for post in session.query(Post):
        # 截断标题到300个字符
        post.new_title = post.title[:300]

    session.commit()

    # 3. 删除旧的title列
    op.drop_column('posts', 'title')

    # 4. 将临时列重命名为title
    op.alter_column('posts', 'new_title', new_column_name='title', nullable=False)

def downgrade():
    # 1. 先添加一个临时列
    op.add_column('posts', sa.Column('old_title', sa.String(200)))

    # 2. 截断过长的标题并迁移到临时列
    bind = op.get_bind()
    Session = sessionmaker(bind=bind)
    session = Session()

    # 这里需要重新定义Post模型，因为现在title是300长度
    class PostDowngrade(sa.ext.declarative.Base):
        __tablename__ = 'posts'
        id = sa.Column(sa.Integer, primary_key=True)
        title = sa.Column(sa.String(300))

    for post in session.query(PostDowngrade):
        # 截断标题到200个字符
        post.old_title = post.title[:200]

    session.commit()

    # 3. 删除新的title列
    op.drop_column('posts', 'title')

    # 4. 将临时列重命名为title
    op.alter_column('posts', 'old_title', new_column_name='title', nullable=False)

应用这个迁移：

alembic upgrade head

5.7 回滚操作

如果发现最新的迁移有问题，可以回滚到上一个版本：

alembic downgrade -1

修复问题后，重新生成并应用迁移。

六、相关资源

  • PyPI地址：https://pypi.org/project/alembic/
  • Github地址：https://github.com/sqlalchemy/alembic
  • 官方文档地址：https://alembic.sqlalchemy.org/

通过本文的介绍，你应该已经掌握了Alembic的基本使用方法和一些高级技巧。Alembic作为一个强大的数据库迁移工具，能够帮助你在项目开发过程中轻松管理数据库结构的变更，保持数据库设计与代码模型的同步。无论是小型项目还是大型应用，Alembic都能为你的数据库迁移提供可靠的支持。{ Environment.NewLine }{ Environment.NewLine }关注我，每天分享一个实用的Python自动化工具。

更多文章