SQLAlchemy で list または dict を update + json_set で更新する方法

• 概要
• 環境
• サンプルコード
• 少し解説
• 最後に

概要

Python の list や dict をそのまま json_set することはできません

MySQL などの世界で使われている JSON 型に変換してあげてから update しましょう

環境

• macOS 11.6.1
• MySQL 8.0.26
• flask-sqlalchemy 2.5.1

サンプルコード

"""flask_sqlalchemyを使ってJSON型をテストするクラス."""
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_marshmallow import Marshmallow

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+mysqldb://root@localhost/test?charset=utf8'  # noqa: E501
db = SQLAlchemy(app)
ma = Marshmallow()


class User(db.Model):
    """Userテーブルのモデル."""

    id = db.Column(db.Integer, primary_key=True)
    profile = db.Column(db.JSON())


class UserResponseSchema(ma.Schema):
    """Userテーブル用のレスポンスデータスキーム."""

    id = ma.String(attribute="id")
    profile = ma.Dict(attribute="profile")


class UserTable():
    """Userテーブルを操作するクラス."""

    def insert(self, profile):
        """profileを登録します."""
        p = User(profile=profile)
        db.session.add(p)
        db.session.commit()

    def select_all(self):
        """全件取得."""
        return User.query.filter().all()

    def upsert(self, id, profile):
        """フィールドがない場合は登録します."""
        user_query = User.query.filter(User.id == id)
        for key, value in profile.items():
            if isinstance(value, list) or isinstance(value, dict):
                # dict or list の場合
                user_query.update(
                    {"profile": db.func.json_set(
                        User.profile,
                        "$." + key,
                        db.func.cast(value, db.JSON))}, synchronize_session='fetch')  # noqa: E501
                db.session.commit()
            else:
                # str や int など構造を持たない値の場合
                user_query.update(
                    {"profile": db.func.json_set(
                        User.profile,
                        "$." + key,
                        value)}, synchronize_session='fetch')
                db.session.commit()


if __name__ == '__main__':
    # レコード取得
    user_table = UserTable()
    records = user_table.select_all()
    ret = UserResponseSchema(many=True).dump(records)

    # レコード追加
    # list の場合
    new_profile = {'age': 10, 'name': 'snowlog', 'items': ["sword", "shield"]}
    user_table.upsert(ret[0]["id"], new_profile)
    records = user_table.select_all()
    ret = UserResponseSchema(many=True).dump(records)
    print(ret)

    # dict の場合
    new_profile = {'age': 10, 'name': 'snowlog', 'items': {"sword": "lv30", "shield": "lv30"}}  # noqa: E501
    user_table.upsert(ret[0]["id"], new_profile)
    records = user_table.select_all()
    ret = UserResponseSchema(many=True).dump(records)
    print(ret)

少し解説

upsert メソッドで更新する値の型をチェックします
list or dict の場合には db.func.cast(value, db.JSON) を使って Python -> MySQL の JSON 型に変換しています

もしこれを行わないで配列や辞書の update + json_set しようとするとシンタックスエラーが発生します

また空の配列や辞書を登録することも可能です

最後に

json_array() や json_object() メソッドを使って MySQL の世界の型に変換してもいいのですがいかんせん Python の配列や辞書のデータをそのまま渡せないので cast を使っています

SQLAlchemy で cast メソッドの使い方はあまり紹介されていないので参考になれば幸いです