2021年7月8日木曜日

SQLAlchemy で JSON 型を扱う方法

概要
環境
テーブル/カラム定義
データの登録
データの取得
データの更新
最後に

概要

MySQL の JSON 型を SQLAlchemy から操作する方法を紹介しますテーブルの定義からデータの登録、取得、更新あたりを試してみましたなお以下のサンプルの SQLAlchemy は Flask-SQLAlchemy 経由で操作しています

環境

macOS 11.4
Python 3.8.3
MySQL 8.0.25
SQLAlchemy 1.3.19

テーブル/カラム定義

db.JSON() を使います

vim schema.py

from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    profile = db.Column(db.JSON())

データの登録

今回は insert するためのクラスを作成しました

vim user_table.py

from schema import User

class UserTable():
    def insert(self, profile):
        p = User(profile=profile)
        db.session.add(p)
        db.session.commit()

これを使って以下のように登録できます Python の dict を使ってそのまま JSON 型のカラムにデータを保存することができます

vim test.py

from user_table import UserTable

user_table = UserTable()
profile = {"name": "hawksnowlog"}
user_table.insert(profile)

データの取得

今回は Mashmallow を使って JSON 型から Python の dict へシリアライズしてます発行したクエリの結果をそのまま Mashmallow に渡せば OK です

vim schema.py

from flask_marshmallow import Marshmallow
from flask_sqlalchemy import SQLAlchemy

db = SQLAlchemy()
ma = Marshmallow()

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    profile = db.Column(db.JSON())

class UserResponseSchema(ma.Schema):
    id = ma.String(attribute="id")
    profile = ma.Dict(attribute="profile")

vim user_table.py

from schema import User

class UserTable():
    def insert(self, profile):
        p = User(profile=profile)
        db.session.add(p)
        db.session.commit()

    def select_all(self):
        return User.query.filter().all()

vim test.py

from user_table import UserTable

user_table = UserTable()
records = parameter_table.select_all()
UserResponseSchema(many=True).dump(records)

データの更新

基本的には登録と同じです一度レコードを取得してそのオブジェクトに対して更新をかけます

vim user_table.py

from schema import User

class UserTable():
    def insert(self, profile):
        p = User(profile=profile)
        db.session.add(p)
        db.session.commit()

    def select_all(self):
        return User.query.filter().all()

    def select_one(self, id):
        return User.query.get(id)

    def update(self, id, profile):
        p = self.select_one(id)
        p.profile = profile
        db.session.add(p)
        db.session.commit()

vim test.py

from user_table import UserTable

user_table = UserTable()
profile = {"name": "updated_hawksnowlog"}
user_table.update(profile)

最後に

基本的なクエリであれば特に詰まることはなくできると思います JSON_EXTRACT や JSON_VALUE を使う場合は今回は紹介していませんが少しハマるかもしれません (SQLAlchemy が対応しているかによる)

また JSON 型を使用する場合は MySQL のバージョンに気をつけましょう

hawksnowlog