概要
ほぼ違いはありませんがそれぞれのサンプルコードを紹介します
基本的には 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 を使うのが良いかなと思います
