初めに
最近携わっているWEBシステムですが、ログインユーザー毎にDBスキーマが異なる仕様で、
「ユーザーログイン時、動的にスキーマ変更してDBアクセス」する必要がありました。
Flask-SQLAlchemyを用いて上記の仕様を実現しようとしたとき、動的にモデルのスキーマを変更する良い方法がなかなか見つからず、１週間ほど悶々としながら悩んでいました。
でも！最近ようやっと解決できたので、あまりスマートではないですが解決方法を書いていきます。
同じように悩んでいる人の手助けになれば幸いです。
検証環境
- Windows10
- PostgreSQL 12.3
- Python 3.6.8
- Flask 1.1.2
- Flask-SQLAlchemy 2.4.3
前提知識
検証するモデル
hogeという同一構造のテーブルが、fooスキーマとbarスキーマに存在するとします。
（以下のコードはpublicスキーマにHogeテーブルを作成するコード）
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] \
= 'postgresql+psycopg2://postgres:postgres@localhost:5432/postgres' # ロカールのデータベース(postgres)に接続
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
class Hoge(db.Model): # public.hogeのモデル
id = db.Column(db.Integer, primary_key=True)
name = db.Column(db.String(80), unique=True, nullable=False)
db.create_all() # hogeテーブルをデータベースに作成
【解決】typeで動的にモデルクラス作成
get_model_dict というスキーマ指定してモデルの中身を返す関数を定義し、スキーマの辞書からtypeを使用して動的にモデルクラスを生成。最終的にスキーマをキーとしたモデルクラスの辞書を作成しました。
利用時にスキーマをキーとしてモデルクラス取得すれば、あとは普通のモデルクラスと同等に利用可能です。
残念な点としては、通常のモデルクラスと比べて書き方が面倒くさいところ。
通常のモデルクラス同じように書きたかったけれどダメでした。。。
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] \
= 'postgresql+psycopg2://postgres:postgres@localhost:5432/postgres' # ロカールのデータベース(postgres)に接続
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
db = SQLAlchemy(app)
# モデルクラスの中身を辞書で返す関数
def get_model_dict(schema):
return {
"__tablename__": 'hoge',
"__table_args__": {"schema": schema},
"id": db.Column(db.Integer, primary_key=True),
"name": db.Column(db.String(80), unique=True, nullable=False),
"__repr__": lambda self: "<Hoge id={}, name={}>".format(self.id, self.name) # __repr__メソッドをlambdaで作成
}
# スキーマの辞書
schema_dict = {1: "foo", 2: "bar"}
# スキーマをキーとしたモデルクラスの辞書を作成
# ※typeで作成するクラス名は被らないように！（被ると怒られた）
HogeDict = {v: type("Hoge{}".format(k), (db.Model, ), get_model_dict(v)) for k, v in schema_dict.items()}
# DBにスキーマ作成
db.session.execute("CREATE SCHEMA foo;")
db.session.execute("CREATE SCHEMA bar;")
db.session.commit()
# DBにhogeテーブル作成
db.create_all()
# ***** 以下動作検証 **********
# hogeテーブルにデータ挿入
db.session.add(HogeDict["foo"](name="I am foo"))
db.session.add(HogeDict["bar"](name="I am bar"))
db.session.commit()
# fooスキーマを検索
HogeDict["foo"].query.all()
# [<Hoge id=1, name=I am foo>]
# barスキーマを検索
HogeDict["bar"].query.all()
# [<Hoge id=1, name=I am bar>]
# 後片付け
db.session.execute("DROP SCHEMA foo, bar CASCADE;")
db.session.commit()
以上です