概要

ほぼ違いはありませんがそれぞれのサンプルコードを紹介します
基本的には type_coerce を使うのが良いかなと思います

環境

macOS 14.2.1
Python 3.11.6
sqlalchemy 2.0.25

テーブル準備

CREATE TABLE user (id int(11) NOT NULL AUTO_INCREMENT, profile json DEFAULT NULL, PRIMARY KEY (id));

INSERT INTO user VALUES (null, '{"name":"hawk"}');
INSERT INTO user VALUES (null, '{"name":"snowlog"}');
INSERT INTO user VALUES (null, '{"name":"hawksnowlog"}');

cast サンプルコード

from typing import Any

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.schema import Column
from sqlalchemy.sql.expression import func
from sqlalchemy.types import JSON, String

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionClass = sessionmaker(engine)
db_session = SessionClass()

Base = declarative_base()


class User(Base):

    __tablename__ = 'user'

    id = Column(String(32), primary_key=True)
    profile = Column(JSON())


class UserTable():

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

    def upsert(self, id: int, profile: dict[str, Any]):
        user_query = db_session.query(User).filter(User.id == id)
        for key, value in profile.items():
            if isinstance(value, list) or isinstance(value, dict):
                user_query.update(
                    {"profile": func.json_set(
                        User.profile,
                        "$." + key,
                        func.cast(value, JSON))}, synchronize_session='fetch')
                db_session.commit()
            else:
                user_query.update(
                    {"profile": func.json_set(
                        User.profile,
                        "$." + key,
                        value)}, synchronize_session='fetch')
                db_session.commit()


if __name__ == '__main__':
    user_table = UserTable()
    user_table.upsert(1, {"name": "hoge", "langs": ["python", "ruby"]})
    user_table.upsert(2, {"name": "fuga", "age": 1})
    records = user_table.select_all()
    for r in records:
        print(r.id)
        print(r.profile)

おまけ func.cast の場合

from typing import Any

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.schema import Column
from sqlalchemy.sql.expression import cast, func
from sqlalchemy.types import JSON, String

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionClass = sessionmaker(engine)
db_session = SessionClass()

Base = declarative_base()


class User(Base):

    __tablename__ = 'user'

    id = Column(String(32), primary_key=True)
    profile = Column(JSON())


class UserTable():

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

    def upsert(self, id: int, profile: dict[str, Any]):
        user_query = db_session.query(User).filter(User.id == id)
        for key, value in profile.items():
            if isinstance(value, list) or isinstance(value, dict):
                user_query.update(
                    {"profile": func.json_set(
                        User.profile,
                        "$." + key,
                        cast(value, JSON))}, synchronize_session='fetch')
                db_session.commit()
            else:
                user_query.update(
                    {"profile": func.json_set(
                        User.profile,
                        "$." + key,
                        value)}, synchronize_session='fetch')
                db_session.commit()


if __name__ == '__main__':
    user_table = UserTable()
    user_table.upsert(1, {"name": "hoge", "langs": ["python", "swift"]})
    user_table.upsert(2, {"name": "fuga", "age": 1})
    records = user_table.select_all()
    for r in records:
        print(r.id)
        print(r.profile)

type_coerce サンプルコード

from typing import Any

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from sqlalchemy.schema import Column
from sqlalchemy.sql.expression import func, type_coerce
from sqlalchemy.types import JSON, String

engine = create_engine("mysql+pymysql://root@localhost/test?charset=utf8mb4")
SessionClass = sessionmaker(engine)
db_session = SessionClass()

Base = declarative_base()


class User(Base):

    __tablename__ = 'user'

    id = Column(String(32), primary_key=True)
    profile = Column(JSON())


class UserTable():

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

    def upsert(self, id: int, profile: dict[str, Any]):
        user_query = db_session.query(User).filter(User.id == id)
        for key, value in profile.items():
            if isinstance(value, list) or isinstance(value, dict):
                user_query.update(
                    {"profile": func.json_set(
                        User.profile,
                        "$." + key,
                        type_coerce(value, JSON))}, synchronize_session='fetch')
                db_session.commit()
            else:
                user_query.update(
                    {"profile": func.json_set(
                        User.profile,
                        "$." + key,
                        value)}, synchronize_session='fetch')
                db_session.commit()


if __name__ == '__main__':
    user_table = UserTable()
    user_table.upsert(1, {"name": "hoge", "langs": ["python", "swift"]})
    user_table.upsert(2, {"name": "fuga", "age": 1})
    records = user_table.select_all()
    for r in records:
        print(r.id)
        print(r.profile)

ちょっと解説

結果の違いはどちらも同じです
どちらも同じように動きます

func に type_coerce はありませんが func から cast はコールできます
公式での type_coerce の説明は「Associate a SQL expression with a particular type, without rendering CAST.」

SQL 側の機能ではなく Python 側の機能だけで型変換を行う感じかなと思います
公式を読む限りでは cast の代替として type_coerce を使うべきだとあるので基本的には type_coerce を使うのが良いかなと思います

参考サイト

https://docs.sqlalchemy.org/en/20/core/sqlelement.html#sqlalchemy.sql.expression.type_coerce

hawksnowlog

2024年1月16日火曜日

SQLAlchemy の cast と type_coerce の使い分けとサンプルコード

概要

環境