Flask SQLAlchemy에서 모델 스키마를 동적으로 변경하는 방법

처음에

최근 종사하고 있는 WEB 시스템입니다만, 로그인 유저마다 DB 스키마가 다른 사양으로,
"사용자 로그인 시 동적으로 스키마를 변경하여 DB 액세스"해야 했습니다.

Flask-SQLAlchemy를 이용해 위의 사양을 실현하려고 했을 때, 동적으로 모델의 스키마를 변경하는 좋은 방법이 좀처럼 발견되지 않고, 1주일 정도 괴로워하면서 고민하고 있었습니다.

하지만! 최근에 드디어 해결할 수 있었으므로, 그다지 스마트하지는 않지만 해결 방법을 써 갑니다.
똑같이 고민하고 있는 사람의 도움이 되면 다행입니다.

검증 환경

Windows10

PostgreSQL 12.3

Python 3.6.8

Flask 1.1.2

Flask-SQLAlchemy 2.4.3

전제 지식

파이썬을 3개월 정도 공부

Flask 튜토리얼 졸업

Flask-SQLAlchemy 빠른 시작 졸업

PosgreSQL에서 스키마와 테이블의 의미를 왠지 아는 사람

검증할 모델

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()

이상입니다