ソムリエでエンジニアのブログ

ワインの事も書きたいけど基本エンジニア用

SQL Alchemy, Alembicでマルチテナント型のデータベース生成とマイグレーションを実行する方法

概要

本業でPythonで処理を書き, ORMとしてSQL Alchemy, マイグレーションの管理をSQL Alchemy, で実施しているプロジェクトがあります。
そのプロジェクトではマルチテナント方式?(一つのユーザーに一つのスキーマーが切られているイメージです。)を採用しており、契約時に新規のスキーマー作成、マイグレーションの実施を行いたいニーズがありました。
ネットには、SQL Alchemy, Alembicを利用してテーブル作成を行う記事は存在したのですが、今回のようなデータベース自体の作成やマイグレーション先をコード上で動的に切り替える方法などは見つかりませんでしたので実装方法の殴り書きとして残しておきます。 (※注意)MariaDBを利用した為、スキーマではなく以後DBを作成と書かせていただきます。

前提条件

通常のマイグレーションは実施できる状態。 alembicのenv.pyで以下のように環境変数からDB関連の値を読み取ってurlを指定しています。

alembic/env.py

from env import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
from alembic import context
import models

... 略

db_url = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
config.set_main_option("sqlalchemy.url", db_url)

... 略

この場合、環境変数で指定したDB_NAMEというデータベースにマイグレーションが実行されます。
この時点では、サーバーに入り手動でDBを作成しalembicのマイグレーション実行コマンドを叩いていました。今回はこれをあるエンドポイントを叩くと引数のDB名でDBを作成しマイグレーションまで実行してくれるように設定していきます。

実装

from env import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT
from sqlalchemy_utils import create_database, database_exists
from sqlalchemy import create_engine
from alembic.config import Config
from alembic import command


def settingDatabase(database_name: str):
    DB_URL = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{database_name}"
    Engine = create_engine(DB_URL, encoding="utf-8")

    if database_exists(DB_URL):
        return False

    create_database(Engine.url)
    alembic_config.set_main_option("sqlalchemy.url", DB_URL)
    command.upgrade(alembic_config, 'head')

    return database_exists(DB_URL)

alembic_config = Config('./alembic.ini')

db.pyでURLを指定するようにしたのでalembicのenvファイルではURLの指定をコメントアウト
これが残っているとdb.pyで指定したURLが上書きされる為うまく動作しません。

alembic/env.py

from env import DB_USER, DB_PASSWORD, DB_HOST, DB_PORT, DB_NAME
from alembic import context
import models

... 略

# db_url = f"mysql+pymysql://{DB_USER}:{DB_PASSWORD}@{DB_HOST}:{DB_PORT}/{DB_NAME}"
# config.set_main_option("sqlalchemy.url", db_url)

... 略

あとは定義したsettingDatabaseメソッドを呼び出せば引数の値を名前にもつDBが作成されマイグレーションも実行されます。

最後に

Python、Alembic、SQL Alchemy全て今回初めて触る技術になりますのでもしかすると見当違いな実装になっているかもしれません。
何か別の方法があればぜひご教授いただけますと幸いです。