概要
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 のバージョンに気をつけましょう
