課題
alembicでテーブル定義のマイグレーションファイルを作る場合に以下の課題があった。
モデル定義ファイルをテーブルごとに分割すると、ファイル名の並び順でマイグレーションの定義が作成される。
要求
依存関係があるとエラーになるので、指定の順序で定義を出力したい。
前提
- modelsフォルダにSQLModelを利用してテーブル定義を作成する
- 「models/__ init__.py」に定義を作成した順番にimportを記載する
models
└ __init__.py
└ user_assign.py
└ user.py
ファイルの並びは上記のようになりmigrationファイルの出力結果も名前の並び替え順になる。
それを、定義の順番である「user → user_assign」に変更して、出力する。
__init__.py
from .user import User
from .user_assign import UserAssign
上記の「__ init__.py」のインポート順で定義を出力する。
alembic/env.pyでmetadataを取得しやすいように、モデルは「sqlmode.SQLModel」を継承しておく。
user.py
from sqlmodel import SQLModel
class User(SQLModel):
... # 具体的な定義
定義
alembic/env.pyに以下を記載します。
デフォルトの記載は省略します。加筆する部分のみ記載します。
alembic/env.py
... デフォルトの定義は省略
# metadataを取得するためにインポート
from sqlmodel import SQLModel
# 各テーブル定義をインポート
from models import *
... デフォルtの定義は省略(「target_metadata = None」を以下のように変更)
target_metadata = SQLModel.metadata
... デフォルトの定義は省略
# ★並び替えの関数を追加
def process_revision_directives(context, revision, directives):
if directives[0].upgrade_ops.ops:
# 定義順になるようにリストを作成(metadata.tablesは定義順になっている)
table_order = list(SQLModel.metadata.tables.keys())
# 既存の操作をソート
sorted_ops = sorted(
directives[0].upgrade_ops.ops,
key=lambda op: (
table_order.index(op.table_name)
if hasattr(op, "table_name") and op.table_name in table_order
else len(table_order)
),
)
# ソートされた操作で置き換え
directives[0].upgrade_ops.ops = sorted_ops
# ★ 並び替えの関数定義ここまで
def run_migrations_offline() -> None:
url = config.get_main_option("sqlalchemy.url")
context.configure(
url=url,
target_metadata=target_metadata,
literal_binds=True,
dialect_opts={"paramstyle": "named"},
process_revision_directives=process_revision_directives, # ★この行を追加
)
with context.begin_transaction():
context.run_migrations()
def run_migrations_online() -> None:
connectable = engine_from_config(
config.get_section(config.config_ini_section, {}),
prefix="sqlalchemy.",
poolclass=pool.NullPool,
)
with connectable.connect() as connection:
context.configure(
connection=connection,
target_metadata=target_metadata,
process_revision_directives=process_revision_directives, # ★この行を追加
)
with context.begin_transaction():
context.run_migrations()
上記の定義後、以下のようにマイグレーションファイルを作成する。
alembic revision --autogenerate -m "create-basic-tables"
alembic/versionsに以下のような順番の定義ファイルが作成される。
バージョン定義ファイル
def upgrade() -> None:
op.create_table('user',...)
op.create_table('user_assign',...)
参考
perplexityに質問して、その内容を実証してみました。
質問:alembicのprocess_revision_directivesでテーブルの順序を指定するには?