SQLAlchemy スキーマを動的に変更する方法

概要

PostgreSQLだとDBとテーブルの間にスキーマが存在しますが、別スキーマの同名テーブルに動的に接続を切り替えたい場合、以下のように関数で囲むことで上手くいきました。

コード例

ORMのモデルクラスを作成し、関数で囲むようにします。

tables/Users.py

from sqlalchemy import Column, VARCHAR, DateTime
from sqlalchemy.orm import declarative_base
from datetime import datetime

Base = declarative_base()


def get_users_orm(schema_name="public"):
    class Users(Base):
        __tablename__ = "users"
        __table_args__ = {"schema": schema_name, "extend_existing": True}

        user_id = Column(VARCHAR(100), primary_key=True, nullable=False)
        name = Column(VARCHAR(100), nullable=False)
        email = Column(VARCHAR(100), nullable=False)
        password = Column(VARCHAR(200), nullable=False)
        created_at = Column(DateTime, default=datetime.now, nullable=False)
        updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, nullable=False)

    return Users

上記は以下のように使用します。

from tables.Users import get_users_orm

# ※sessionの取得は省略

# privateスキーマのusersテーブルのnameカラムが"スキーマ太郎"のレコードを取得
users_orm = get_users_orm("private")
session.query(users_orm).filter(users_orm.name == "スキーマ太郎").first()

詳細

schema_nameに切り替え対象のスキーマを設定します。
また、上記関数を呼び出すごとに、テーブルが再定義されるようなので、__table_args__に

"extend_existing": True

を設定し、再定義を許可しています。

補足

最初はインスタンス化時にclassの引数へ設定してみたのですが、スキーマは切り替わりませんでした。
調べたらSQLAlchemyでは、ALTER TABLE文をサポートしていないようです。

※pytestで上記関数を使う場合、再定義されてしまっていますよ、というwarningが出たので、pytest.iniなどでエラーメッセージを非表示にする必要があります。