【Python】マイグレーションツールAlembicを触ってみる

SQLAlchemyで使えるマイグレーションツールAlembicの使い方を調べました。
自分の中でベストな構成(今のところ)が完成したので残しておきます。

https://alembic.sqlalchemy.org/en/latest/

# 準備

```bash
$ python3.8 --version
Python 3.8.13

# venv
$ python3.8 -m venv .venv
$ source .venv/bin/activate

# パッケージインストール
$ pip install alembic pymysql SQLAlchemy

# ディレクトリ作成
$ mkdir -p db models
$ touch db/__init__.py models/__init__.py

# データベース作成
$ MYSQL_PWD="XXXXXXXXX" mysql -u user -h somedb.example.com  -e "create database alembic_tutorial"
```

# ディレクトリ構成

```
  alembic-tutorial/
  | alembic/                      # alembicリソースが格納されるディレクトリ
  | | versions/                   # マイグレーションファイル格納ディレクトリ
  | | README
  | | env.py                      # DBとの接続情報などを設定するスクリプト
  | | script.py.mako
  | db/
  | | __init__.py
  | | base.py                     # Baseクラス、model読み込み
  | | base_class.py               # Baseクラス定義
  | | db.py                       # engine生成session取得関数定義
  | models/                       # モデル格納ディレクトリ
  | | __init__.py
  | | user.py
  | alembic.ini                   # alembicの設定ファイル
  | main.py                       # テーブルを操作するスクリプト
  ```

# モデルの作成
マイグレーション対象のモデルを作成します。

```bash
# ファイルの作成
$ touch db/base.py db/base_class.py db/db.py models/user.py
```

モデルが継承するベースクラス
```db/base_class.py
from sqlalchemy.orm.decl_api import declarative_base
Base = declarative_base()
```

先程の **Base** を継承してモデルを実装
```models/user.py
from datetime import datetime
from sqlalchemy import Boolean, Column, Integer, String
from sqlalchemy.orm import relationship
from sqlalchemy.sql.sqltypes import DateTime

from db.base_class import Base

class User(Base):
    __tablename__ = "users"
    __table_args__ = {'mysql_engine':'InnoDB', 'mysql_charset':'utf8mb4','mysql_collate':'utf8mb4_bin'}
    

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(255, collation="utf8mb4_bin"), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True, nullable=False)
    created = Column(DateTime, default=datetime.now, nullable=False)
    updated = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False)
```

DBとの接続を管理する実装
```db/db.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

db_user = "user"
db_passwd = "XXXXXXXXX"
db_host = "somedb.example.com"
db_name = "alembic_tutorial"
DB_URL = f'mysql+pymysql://{db_user}:{db_passwd}@{db_host}/{db_name}?charset=utf8mb4'

engine = create_engine(DB_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=True, bind=engine)

def get_db():
    return SessionLocal()
```

定義したモデルをまとめてimportする実装
```db/base.py
from db.base_class import Base
# すべてのモデルをimport (alembicにモデル情報を引き渡すために必要です)
from models.user import User
```


# alembicプロジェクトの初期化

```bash
$ mkdir alembic

# 初期化 (alembic.ini や alembic/env.py などのリソースが生成されます。)
$ alembic init alembic

$ ls alembic.ini
alembic.ini
$ ls alembic/
env.py  README  script.py.mako  versions
```


# alembic/env.pyの修正
[Alembic  | Auto Generating Migrations](https://alembic.sqlalchemy.org/en/latest/autogenerate.html)

alembicを先程作成したモデルに紐付けるために、 alembic/env.py に以下のの修正を行います。

1. モデルの情報をalembicに引き渡す
`target_metadata` に db/base.py からimportした `base.Base.metadata` を設定します。
<font color="orange">※ モデル情報をtarget_metadataに引き渡すために、`db/base.py` で、すべてのモデルをインポートしています。</font>
1. alembicのDB接続情報を db/db.py の `DB_URL` に修正
接続情報は alembic.ini の  `sqlalchemy.url` に定義されていますが、環境変数などで動的に設定したいので、 alembic/env.py で db/db.py の `DB_URL` を読み込みます。
1. マイグレーションスクリプト生成時に列タイプの変更を検知できるようにする
 `context.configure` の引数に `compare_type=True` を設定します。
[configure.params.compare_type | Alembic](https://alembic.sqlalchemy.org/en/latest/api/runtime.html#alembic.runtime.environment.EnvironmentContext.configure.params.compare_type)

```alembic/env.py
from logging.config import fileConfig

from sqlalchemy import engine_from_config
from sqlalchemy import pool

from alembic import context

config = context.config

if config.config_file_name is not None:
    fileConfig(config.config_file_name)

# target_metadata = None # コメントアウト

from db.db import DB_URL # DBの接続情報をimport
from db import base # モデルクラスの読み込み
target_metadata = base.Base.metadata # db/base.pyで定義したBaseクラスのメタデータを使用

def run_migrations_offline() -> None:
    # url = config.get_main_option("sqlalchemy.url") # コメントアウト
    context.configure(
        url=DB_URL, # db/db.pyからimportした接続情報を利用する
        target_metadata=target_metadata,
        literal_binds=True,
        dialect_opts={"paramstyle": "named"},
        compare_type=True, # マイグレーションスクリプト生成時に列タイプの変更を検知するオプションを有効化
    )

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


def run_migrations_online() -> None:
    conf = config.get_section(config.config_ini_section) # alembic.iniの読み込み
    conf["sqlalchemy.url"] = DB_URL # sqlalchemy.urlをdb/db.pyからimportした接続情報で上書き
    connectable = engine_from_config(
        conf,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )

    with connectable.connect() as connection:
        context.configure(
            connection=connection,
            target_metadata=target_metadata,
            compare_type=True, # マイグレーションスクリプト生成時に列タイプの変更を検知するオプションを有効化
        )

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


if context.is_offline_mode():
    run_migrations_offline()
else:
    run_migrations_online()
```


# マイグレーションファイル生成
モデルの定義からマイグレーションファイルを自動生成します。

```bash
# モデル定義からマイグレーションスクリプトを自動生成
# -m "COMMENT" : '_' つなぎのファイル名になります
alembic revision --autogenerate -m "create initial table"
```

生成されるマイグレーションスクリプト
```alembic/versions/4a8221af54a3_create_initial_table.py
"""create initial table

Revision ID: 4a8221af54a3
Revises: 
Create Date: 2022-10-04 14:08:42.625161

"""
from alembic import op
import sqlalchemy as sa


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


def upgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.create_table('users',
    sa.Column('id', sa.Integer(), nullable=False),
    sa.Column('username', sa.String(length=255, collation='utf8mb4_bin'), nullable=False),
    sa.Column('hashed_password', sa.String(length=255), nullable=False),
    sa.Column('is_active', sa.Boolean(), nullable=False),
    sa.Column('created', sa.DateTime(), nullable=False),
    sa.Column('updated', sa.DateTime(), nullable=False),
    sa.PrimaryKeyConstraint('id'),
    mysql_charset='utf8mb4',
    mysql_collate='utf8mb4_bin',
    mysql_engine='InnoDB'
    )
    op.create_index(op.f('ix_users_id'), 'users', ['id'], unique=False)
    op.create_index(op.f('ix_users_username'), 'users', ['username'], unique=True)
    # ### end Alembic commands ###


def downgrade() -> None:
    # ### commands auto generated by Alembic - please adjust! ###
    op.drop_index(op.f('ix_users_username'), table_name='users')
    op.drop_index(op.f('ix_users_id'), table_name='users')
    op.drop_table('users')
    # ### end Alembic commands ###

```

DBにマイグレーションバージョン管理テーブルが生成されます。

```bash
$ MYSQL_PWD="XXXXXXXXX" mysql -u user -h somedb.example.com  alembic_tutorial -e "show tables"
+----------------------------+
| Tables_in_alembic_tutorial |
+----------------------------+
| alembic_version            |
+----------------------------+
```

## --autogenerateで自動検出される変更にはルールがあるらしい
[Autogenerate Detect](https://alembic.sqlalchemy.org/en/latest/autogenerate.html?highlight=compare_server_default#what-does-autogenerate-detect-and-what-does-it-not-detect)

- デフォルトで検出される変更
    - テーブルの追加、削除
    - 列の追加、削除
    - 列のNULL可能ステータスの変更
    - インデックスの基本的な変更と明示的に名前が付けられたユニーク製薬
    - 外部キー制約の基本的な変更
- `compare_type=True` の設定で検出される項目
    - 列タイプの変更
- `compare_server_default=True` の設定で検出される項目
    - サーバーのデフォルトの変更？
- 検出できない変更
    - 列名の変更
    - 匿名で名前が付けられた制約
    - Enum

# マイグレーション実行

```bash
# 最新のバージョンまでマイグレーション
$ alembic upgrade head 

# alembic_versionテーブルに適用されたバージョンが登録されます。
$ MYSQL_PWD="XXXXXXXXX" mysql -u user -h somedb.example.com  alembic_tutorial -e "select * from alembic_version"
+--------------+
| version_num  |
+--------------+
| 4a8221af54a3 |
+--------------+
```

# 作成したテーブルに値を登録してみる

userを登録して、表示して、削除するスクリプトを実装してみます。
```main.py
from db import base
from db import db
from pprint import pprint

if __name__ == "__main__":
    # ジェネレータから要素を取得
    session = db.get_db()
    user_name = "user1"

    # user登録
    user1 = base.User(
        username = user_name,
        hashed_password = "1234567890",
        is_active = True,
    )

    session.add(user1)
    session.commit()

    # ユーザーの取得
    rows = session.query(base.User).offset(0).limit(10).all()
    for row in rows:
        pprint(row.__dict__)

    # ユーザーの削除
    user = session.query(base.User).filter(base.User.username == user_name).first()
    session.delete(user)
    session.commit()
```


```bash
$ python main.py
{'_sa_instance_state': <sqlalchemy.orm.state.InstanceState object at 0x7fb04bd84580>,
 'created': datetime.datetime(2022, 10, 4, 14, 48, 10),
 'hashed_password': '1234567890',
 'id': 4,
 'is_active': True,
 'updated': datetime.datetime(2022, 10, 4, 14, 48, 10),
 'username': 'user1'}
```

# モデルの追加
モデルを追加したい場合は、モデルファイルの追加と `db/base.py` への import文の追記が必要になります。

```models/item.py
from datetime import datetime
from sqlalchemy import Column, Integer, String
from sqlalchemy.sql.sqltypes import DateTime
from sqlalchemy.dialects.mysql import MEDIUMTEXT

from db.base_class import Base

class Item(Base):
    __tablename__ = "items"
    __table_args__ = {'mysql_engine':'InnoDB', 'mysql_charset':'utf8mb4','mysql_collate':'utf8mb4_bin'}
    
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False)
    content = Column(MEDIUMTEXT)
    created = Column(DateTime, default=datetime.now, nullable=False)
    updated = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False)
```

alembicにモデルを認識させるため、 models/item.py をインポートします。
```db/base.py
from db.base_class import Base
from models.user import User
from models.item import Item  # 追加
```

マイグレーション実行
```bash
# マイグレーションスクリプト生成
$ alembic revision --autogenerate -m "add items table"

# マイグレーション実行
$ alembic upgrade head
```

# マイグレーションコマンドまとめ
https://alembic.sqlalchemy.org/en/latest/tutorial.html#running-our-first-migration
```bash
# 最新のバージョンまでマイグレーション
alembic upgrade head 

# 一番最初までロールバック
alembic downgrade base

# マイグレーション履歴の確認
alembic history -v

# 次のバージョンにマイグレーション
alembic upgrade +1

# 前のバージョンにロールバック
alembic downgrade -1
```


# 参考

- [Alembic](https://alembic.sqlalchemy.org/en/latest/)