Alembic是一个轻量级的数据库迁移工具，可与SQLAlchemy Database Toolkit for Python一起使用。

Alembic使用SQLAlchemy作为底层引擎。

安装

1
$ pip install alembic

使用

初始化

1
$ alembic init alembic

将会在当前目录下生成一个alembic的目录，像这样：

1
2
3
4
5
6
|--alembic\
|  |--versions\
|  |--env.py
|  |--README
|  |--script.py.mako
|--alembic.ini

创建迁移脚本

1
$ alembic revision -m "first revision"

执行后，将会在versions下创建一个python脚本: 95cd2fc4eacd_first_revision.py

在生成的对应版本的脚本中实现upgrade和downgrade

1
2
3
4
5
6
7
8
9
10
11
12
13
def upgrade():
    op.create_table("user",
        sa.Column("id", sa.Integer(), nullable=False),
        sa.Column("full_name", sa.String(), nullable=True),
        sa.Column("email", sa.String(), nullable=True),
        sa.Column("hashed_password", sa.String(), nullable=True),
        sa.Column("is_active", sa.Boolean(), nullable=True),
        sa.Column("is_superuser", sa.Boolean(), nullable=True),
        sa.PrimaryKeyConstraint("id")
    )
    op.create_index(op.f("ix_user_email"), "user", ["email"], unique=True)
    op.create_index(op.f("ix_user_full_name"), "user", ["full_name"], unique=False)
    op.create_index(op.f("ix_user_id"), "user", ["id"], unique=False)

在upgrade中创建了用户表以及对应的索引，则需要在downgrade中删除对应的表和索引

1
2
3
4
5
def downgrade():
    op.drop_index(op.f("ix_user_id"), table_name="user")
    op.drop_index(op.f("ix_user_full_name"), table_name="user")
    op.drop_index(op.f("ix_user_email"), table_name="user")
    op.drop_table("user")

由于是单独测试alembic, 需要在alembic.ini中修改数据库连接。

1
sqlalchemy.url = postgresql://postgres:TopLinker0510@localhost/test

执行

1
2
3
4
$ alembic upgrade head
INFO  [alembic.runtime.migration] Context impl PostgresqlImpl.
INFO  [alembic.runtime.migration] Will assume transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> 95cd2fc4eacd, first revision

自动生成

一般地，我们会根据Model自动生成迁移脚本。

  1. 修改env.py, 导入当前路径到path中

    1
    2
    3
    import os
    import sys
    sys.path.append(os.path.dirname(os.path.abspath(__file__)) +"/../")

  2. 编写Model

    base_class.py

    1
    2
    3
    from sqlalchemy.ext.declarative import declarative_base

    Base = declarative_base()

user.py

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
from typing import TYPE_CHECKING
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import relationship
from db.base_class import Base

if TYPE_CHECKING:
 from .item import Item

class User(Base):
    __tablename__ = "sys_user"
    
    id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    is_active = Column(Boolean(), default=True)
    is_superuser = Column(Boolean(), default=False)
    items = relationship("Item", back_populates="owner")

  1. 修改env.py，设置target_metadata

    1
    2
    from db import Base
    target_metadata = Base.metadata

  2. 执行命令，生成脚本

    1
    $ alembic revision --autogenerate -m "first revision"

  3. 迁移

    1
    $ alembic upgrade head

    详细代码在这里

Reference