SQLAlchemyで使えるマイグレーションツールAlembicの使い方を調べました。
自分の中でベストな構成(今のところ)が完成したので残しておきます。
準備
$ 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 # テーブルを操作するスクリプト
モデルの作成
マイグレーション対象のモデルを作成します。
# ファイルの作成
$ 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プロジェクトの初期化
$ 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
alembicを先程作成したモデルに紐付けるために、 alembic/env.py に以下のの修正を行います。
- モデルの情報をalembicに引き渡す
target_metadataに db/base.py からimportしたbase.Base.metadataを設定します。
※ モデル情報をtarget_metadataに引き渡すために、db/base.pyで、すべてのモデルをインポートしています。 - alembicのDB接続情報を db/db.py の
DB_URLに修正
接続情報は alembic.ini のsqlalchemy.urlに定義されていますが、環境変数などで動的に設定したいので、 alembic/env.py で db/db.py のDB_URLを読み込みます。 - マイグレーションスクリプト生成時に列タイプの変更を検知できるようにする
context.configureの引数にcompare_type=Trueを設定します。
configure.params.compare_type | Alembic
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()
マイグレーションファイル生成
モデルの定義からマイグレーションファイルを自動生成します。
# モデル定義からマイグレーションスクリプトを自動生成
# -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にマイグレーションバージョン管理テーブルが生成されます。
$ MYSQL_PWD="XXXXXXXXX" mysql -u user -h somedb.example.com alembic_tutorial -e "show tables"
+----------------------------+
| Tables_in_alembic_tutorial |
+----------------------------+
| alembic_version |
+----------------------------+
--autogenerateで自動検出される変更にはルールがあるらしい
- デフォルトで検出される変更
- テーブルの追加、削除
- 列の追加、削除
- 列のNULL可能ステータスの変更
- インデックスの基本的な変更と明示的に名前が付けられたユニーク製薬
- 外部キー制約の基本的な変更
-
compare_type=Trueの設定で検出される項目- 列タイプの変更
-
compare_server_default=Trueの設定で検出される項目- サーバーのデフォルトの変更？
- 検出できない変更
- 列名の変更
- 匿名で名前が付けられた制約
- Enum
マイグレーション実行
# 最新のバージョンまでマイグレーション
$ 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()
$ 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 # 追加
マイグレーション実行
# マイグレーションスクリプト生成
$ alembic revision --autogenerate -m "add items table"
# マイグレーション実行
$ alembic upgrade head
マイグレーションコマンドまとめ
# 最新のバージョンまでマイグレーション
alembic upgrade head
# 一番最初までロールバック
alembic downgrade base
# マイグレーション履歴の確認
alembic history -v
# 次のバージョンにマイグレーション
alembic upgrade +1
# 前のバージョンにロールバック
alembic downgrade -1