alembicでmigrationファイルを１つにまとめる

Web開発

どうも皆さん、こんにちは！

最近、alembicコマンドを使用してマイグレーションファイルを作成する機会が多いのですが。

もしマイグレーションファイルが増えすぎて1つにまとめたい場合はどうするのかといったことが疑問に思ったので、今回それについて記載します。

alembicでのsquashマイグレーション方法

Djangoでは、manage.py squashmigrations というコマンドがあり、これで1つのマイグレーションファイルにまとめてくれます。

しかし、alembicでは単にモデルとDBの差分をみてマイグレーションスクリプトを作成する自動生成コマンドを使用するだけみたいです。

alembic revision --autogenerate

なので、やり方としては。

  1. 空のDBを用意する
  2. alembicのコンフィグを一時的に用意したDBに向ける
  3. alembic revision –autogenerateコマンドを実行する
  4. alembicのコンフィグをもとのDBに戻す
  5. 元のDBのalembic_revisionを最新のものに更新する

となります。

実際にマイグレーションをやってみる

最初は、以下のように一番最初のマイグレーションのみ実行してある状態。

# alembic current
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
fb1001010ac8 (head)

モデルを変更して差分をつくる

なんでもいいのですが、今回はUserが複数のCompanyにも所属するように変更します。

# __init__.py
from .company_user_map import company_user_map_table
from .item import Item
from .user import User
from .company import Company

# company.py
from typing import TYPE_CHECKING

from sqlalchemy import (
    Boolean, Column, Integer, String,
    DateTime, func
    )
from sqlalchemy.orm import relationship

from app.db.base_class import Base

from app.models import company_user_map_table

if TYPE_CHECKING:
    from .user import User  # noqa: F401


class Company(Base):
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100), index=True)
    is_active = Column(Boolean(), default=True)
    created_at = Column(
        DateTime,
        server_default=func.now()
        )
    updated_at = Column(
        DateTime,
        server_default=func.now(),
        onupdate=func.now()
        )

    users = relationship(
        "User",
        secondary=company_user_map_table,
        back_populates="companies"
        )

# user.py
from typing import TYPE_CHECKING

from sqlalchemy import (
    Boolean, Column, Integer, String,
    DateTime, func
    )
from sqlalchemy.orm import relationship

from app.db.base_class import Base

from app.models import company_user_map_table

if TYPE_CHECKING:
    from .item import Item  # noqa: F401
    from .company import Company  # noqa: F401


class User(Base):
    id = Column(Integer, primary_key=True, index=True)
    full_name = Column(String(100), index=True)
    email = Column(String(100), unique=True, index=True, nullable=False)
    hashed_password = Column(String(100), nullable=False)
    is_active = Column(Boolean(), default=True)
    is_superuser = Column(Boolean(), default=False)
    created_at = Column(
        DateTime,
        server_default=func.now()
        )
    updated_at = Column(
        DateTime,
        server_default=func.now(),
        onupdate=func.now()
        )

    companies = relationship(
        "Company",
        secondary=company_user_map_table,
        back_populates="users"
        )
    items = relationship("Item", back_populates="owner")

# company_user_map.py
from sqlalchemy import (Table, ForeignKey, Column, Integer)

from app.db.base_class import Base


company_user_map_table = Table(
    'company_user_map',
    Base.metadata,
    Column('company_id', Integer, ForeignKey('company.id')),
    Column('user_id', Integer, ForeignKey('user.id'))
    )

※ item.py は関係ないので変更していないです。

マイグレーションファイルを追加する

# alembic revision --autogenerate -m "Updated relationship between User and Company"
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'company_user_map'
INFO  [alembic.autogenerate.compare] Detected removed foreign key (company_id)(id) on table user
INFO  [alembic.autogenerate.compare] Detected removed column 'user.company_id'
  Generating
  <PATH>/alembic/versions/9b0f1c1cc286_updated_relationship_between_user_and_.py
  ...  done

ファイルの中身はこんな感じです。

# 9b0f1c1cc286_updated_relationship_between_user_and_.py
"""Updated relationship between User and Company

Revision ID: 9b0f1c1cc286
Revises: fb1001010ac8
Create Date: 2020-XX-XX XX:XX:XX.XXXXXX

"""
from alembic import op
import sqlalchemy as sa
from sqlalchemy.dialects import mysql

# revision identifiers, used by Alembic.
revision = '9b0f1c1cc286'
down_revision = 'fb1001010ac8'
branch_labels = None
depends_on = None


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        'company_user_map',
        sa.Column('company_id', sa.Integer(), nullable=True),
        sa.Column('user_id', sa.Integer(), nullable=True),
        sa.ForeignKeyConstraint(['company_id'], ['company.id'], ),
        sa.ForeignKeyConstraint(['user_id'], ['user.id'], )
    )
    op.drop_constraint('user_ibfk_1', 'user', type_='foreignkey')
    op.drop_column('user', 'company_id')
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.add_column('user', sa.Column('company_id', mysql.INTEGER(display_width=11), autoincrement=False, nullable=False))
    op.create_foreign_key('user_ibfk_1', 'user', 'company', ['company_id'], ['id'])
    op.drop_table('company_user_map')
    # ### end Alembic commands ###

マイグレーションファイルができたので、更新してみます。

# alembic upgrade head
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade fb1001010ac8 -> 9b0f1c1cc286, Updated relationship between User and Company

次からが本番です。

空のDBを用意する

mysql> create database sampledb02;
mysql> GRANT ALL PRIVILEGES ON sampledb02.* TO 'test'@'192.168.XX.XX';

alembicのコンフィグを一時的に用意したDBに向ける

env.pyで設定してあるDBを一時的に入れ替える。

def get_url():
    ...
    db = os.getenv("MYSQL_DATABASE", "sampledb02")
    ...

上記を変更した後、以下コマンドで現在のマイグレーションのversionを確認してみる。

# alembic current
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.

初期状態に戻っているのでOK。

ここで、既存のマイグレーションファイルは削除する。

※gitでversion管理していて、いつでも戻せる状態であることを想定。

alembic revision –autogenerateコマンドを実行する

削除後、新しいマイグレーションファイルを作成する。

# alembic revision --autogenerate -m "Squash migrations"
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.autogenerate.compare] Detected added table 'company'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_company_id' on '['id']'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_company_name' on '['name']'
INFO  [alembic.autogenerate.compare] Detected added table 'user'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_email' on '['email']'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_full_name' on '['full_name']'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_user_id' on '['id']'
INFO  [alembic.autogenerate.compare] Detected added table 'company_user_map'
INFO  [alembic.autogenerate.compare] Detected added table 'item'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_item_id' on '['id']'
INFO  [alembic.autogenerate.compare] Detected added index 'ix_item_title' on '['title']'
  Generating <PATH>/alembic/versions/ea1e4bc52498_squash_migrations.py ...  done

以下、出力されたマイグレーションファイル。

"""Squash migrations

Revision ID: ea1e4bc52498
Revises:
Create Date: 2020-XX-XX XX:XX:XX.XXXXXX

"""
from alembic import op
import sqlalchemy as sa


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


def upgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table(
        'company',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('name', sa.String(length=100), nullable=True),
        sa.Column('is_active', sa.Boolean(), nullable=True),
        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=True),
        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=True),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_company_id'), 'company', ['id'], unique=False)
    op.create_index(op.f('ix_company_name'), 'company', ['name'], unique=False)
    op.create_table(
        'user',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('full_name', sa.String(length=100), nullable=True),
        sa.Column('email', sa.String(length=100), nullable=False),
        sa.Column('hashed_password', sa.String(length=100), nullable=False),
        sa.Column('is_active', sa.Boolean(), nullable=True),
        sa.Column('is_superuser', sa.Boolean(), nullable=True),
        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=True),
        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), 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)
    op.create_table(
        'company_user_map',
        sa.Column('company_id', sa.Integer(), nullable=True),
        sa.Column('user_id', sa.Integer(), nullable=True),
        sa.ForeignKeyConstraint(['company_id'], ['company.id'], ),
        sa.ForeignKeyConstraint(['user_id'], ['user.id'], )
    )
    op.create_table(
        'item',
        sa.Column('id', sa.Integer(), nullable=False),
        sa.Column('title', sa.String(length=100), nullable=True),
        sa.Column('description', sa.String(length=1000), nullable=True),
        sa.Column('owner_id', sa.Integer(), nullable=True),
        sa.Column('created_at', sa.DateTime(), server_default=sa.text('now()'), nullable=True),
        sa.Column('updated_at', sa.DateTime(), server_default=sa.text('now()'), nullable=True),
        sa.ForeignKeyConstraint(['owner_id'], ['user.id'], ),
        sa.PrimaryKeyConstraint('id')
    )
    op.create_index(op.f('ix_item_id'), 'item', ['id'], unique=False)
    op.create_index(op.f('ix_item_title'), 'item', ['title'], unique=False)
    # ### end Alembic commands ###


def downgrade():
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_item_title'), table_name='item')
    op.drop_index(op.f('ix_item_id'), table_name='item')
    op.drop_table('item')
    op.drop_table('company_user_map')
    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')
    op.drop_index(op.f('ix_company_name'), table_name='company')
    op.drop_index(op.f('ix_company_id'), table_name='company')
    op.drop_table('company')
    # ### end Alembic commands ###

ここで、仮のDBに対してマイグレーションファイルを実行して、テーブルが作成されるのかテストしてみる。

# alembic upgrade head
INFO  [alembic.runtime.migration] Context impl MySQLImpl.
INFO  [alembic.runtime.migration] Will assume non-transactional DDL.
INFO  [alembic.runtime.migration] Running upgrade  -> ea1e4bc52498, Squash migrations

alembicのコンフィグをもとのDBに戻す

ここで元のDBへ設定を戻す。

def get_url():
    ...
    db = os.getenv("MYSQL_DATABASE", "sampledb")
    ...

元のDBのalembic_revisionを最新のものに更新する

alembicのマイグレーションversionが元のままなので、これを新しく作成したマイグレーションIDへ更新する。

mysql> select * from alembic_version;
+--------------+
| version_num  |
+--------------+
| 9b0f1c1cc286 |
+--------------+
mysql> update alembic_version set version_num = "ea1e4bc52498";
mysql> select * from alembic_version;
+--------------+
| version_num  |
+--------------+
| ea1e4bc52498 |
+--------------+

以上で完了です。

スポンサーリンク
タイトルとURLをコピーしました