Skip to main content

Alembic 学习

Installation and initiate​

pip install alembic
pip install python-decouple # To load environment variables

# Setup project
alembic init migrations

Config​

# A generic, single database configuration.
[application]
config_file_name = app_conf.py

[alembic]
# path to migration scripts
script_location = foo.bar:migrations

# template used to generate migration files
# file_template = %%(rev)s_%%(slug)s

# set to 'true' to run the environment during
# the 'revision' command, regardless of autogenerate
# revision_environment = false

# Logging configuration
[loggers]
keys = root,sqlalchemy,alembic

[handlers]
keys = console

[formatters]
keys = generic

[logger_root]
level = WARN
handlers = console
qualname =

[logger_sqlalchemy]
level = WARN
handlers =
qualname = sqlalchemy.engine

[logger_alembic]
level = INFO
handlers =
qualname = alembic

[handler_console]
class = StreamHandler
args = (sys.stderr,)
level = NOTSET
formatter = generic

[formatter_generic]
format = %(levelname)-5.5s [%(name)s] %(message)s
datefmt = %H:%M:%S
.env
DATABASE_URL=postgresql://postgres:password@localhost:5432/databasename
env.py
from sqlmodel import SQLModel
from decouple import config

# ...
# Load Database_URL from .env file
DATABASE_URL = config("DATABASE_URL")
# ...
config.set_main_option("sqlalchemy.url", DATABASE_URL)
# ...
# Add 'autogenerate' migration support
from models import Hero
target_metadata = SQLModel.metadata
script.py.mako
# ...
# import sqlalchemy as sa
import sqlmodel
# ${imports if imports else ""}
# ...

Generate migration files​

# Create migration files
alembic revision # Wrong command
alembic revision --autogenerate -m "<migraion_name>"
alembic revision --autogenerate -m "Create hero table"


Upgrade and downgrade migration files​

alembic upgrade head

alembic downgrade head-1
alembic downgrade <commit>

History and info​

alembic history
alembic show <commit>

Migration file example​

"""Create User table

Revision ID: 1234
Revises:
Create Date: 2023-07-20 15:30:00.000000

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic
revision = '1234'
down_revision = None
branch_labels = None
depends_on = None


def upgrade():
    # 创建表操作
    op.create_table(
        'user',
        sa.Column('id', sa.Integer(), nullable=False, primary_key=True),
        sa.Column('email', sa.String(length=255), nullable=False, unique=True),
        sa.Column('hashed_password', sa.String(length=255), nullable=False),
        sa.Column('is_active', sa.Boolean(), server_default='TRUE', nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False)
    )

    # 如果需要创建索引
    op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True)


def downgrade():
    # 回滚操作 - 删除表
    op.drop_table('user')
"""Drop User table

Revision ID: 5678
Revises: 1234
Create Date: 2023-07-21 10:15:00.000000

"""
from alembic import op
import sqlalchemy as sa


# revision identifiers, used by Alembic
revision = '5678'
down_revision = '1234'
branch_labels = None
depends_on = None


def upgrade():
    # 删除表操作
    op.drop_table('user')


def downgrade():
    # 回滚操作 - 重新创建表
    op.create_table(
        'user',
        sa.Column('id', sa.Integer(), nullable=False, primary_key=True),
        sa.Column('email', sa.String(length=255), nullable=False, unique=True),
        sa.Column('hashed_password', sa.String(length=255), nullable=False),
        sa.Column('is_active', sa.Boolean(), server_default='TRUE', nullable=False),
        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False),
        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=False)
    )
    op.create_index(op.f('ix_user_email'), 'user', ['email'], unique=True)

Resource​