SQLAlchemy クエリ実行結果のObjectを辞書(dict) に変換 - Oboe吹きプログラマの黙示録
に続けて、今度は JSON にする変換する方法、
dict を求めてそのまま、json_dumps の実行では、DBの型、DATETIME や TIMESTAMP に
対応できなくなるので、
JSONシリアライズで、datetime に注意する - Oboe吹きプログラマの黙示録
を思い出して、クエリ結果をマッピングクラスの方に、to JSON のメソッドを用意する。
SQLAlchemy クエリ実行結果のObjectを辞書(dict) に変換 - Oboe吹きプログラマの黙示録
の時で使用した ItemWork クラスを例にすると、以下のように辞書(dict) への変換メソッドを持つだけでなく
さらに、JSONへの変換メソッドを持たせる
import sqlalchemy as db from sqlalchemy.orm import declarative_base from sqlalchemy import inspect from sqlalchemy.ext.hybrid import hybrid_property from datetime import date, datetime import json class Itemwork(declarative_base()): __tablename__ = 'itemwork' id = db.Column(db.INT, nullable=False, primary_key=True) user_name = db.Column(db.String(120), nullable=False) task_name = db.Column(db.String(60), nullable=False) point = db.Column(db.INT, nullable=True) create_time = db.Column(db.TIMESTAMP, nullable=False) def __init__(self,id:int, user_name:str, task_name:str, point:int, create_time=datetime.now()): self.id = id self.user_name = user_name self.task_name = task_name self.point = point self.create_time = create_time def to_dict(self) -> {}: dict_ = {} for key in self.__mapper__.c.keys(): if not key.startswith('_'): dict_[key] = getattr(self, key) for key, prop in inspect(self.__class__).all_orm_descriptors.items(): if isinstance(prop, hybrid_property): dict_[key] = getattr(self, key) return dict_ def to_json(self) -> str: return json.dumps(self.to_dict(), default=lambda o: o.strftime('%Y-%m-%d %H:%M:%S') if isinstance(o, datetime) else o.strftime('%Y-%m-%d') if isinstance(o, date) else o.__dict__, indent=2)
クエリ結果から直ぐにJSON文字列を得ることができる
res = session.query(Itemwork).from_statement(text(sql).params(id=2)).first() itemjson = res.to_json()