sqlalchemy 查询结果集 Json序列化
百度搜出来的全是千篇一律抄来抄去的文章，而且我感觉都是搬运的这一篇 Stack Overflow上面的提问 https://stackoverflow.com/questions/5022066/how-to-serialize-sqlalchemy-result-to-json
但遗憾的是，我感觉对我而言都不够优雅。
我的环境
我使用的FastAPI + sqlalchemy 但是没有使用model，直接上SQL。
# db 是 sqlalchemy会话对象
sql = 'SELECT * FROM site_info'
site_info = db.execute(sql).fetchall()
查询到的site_info是一个包含 sqlalchemy ResultProxy的列表
如何优雅的JSON序列化？
众所周知，python json库
import json
json.dumps() # 序列化 返回python中的str类型
json.loads() # 反序列化 返回python中的dict类型
对我而言，我对FastAPI返回值，再次封装了一层，所以我得还原成dict, 再由FastAPI帮我序列化响应。 所以我是这样封装的。
import json
import decimal
import datetime
def _alchemy_encoder(obj):
"""
处理序列化中的时间和小数
:param obj:
:return:
"""
if isinstance(obj, datetime.date):
return obj.strftime("%Y-%m-%d %H:%M:%S")
elif isinstance(obj, decimal.Decimal):
return float(obj)
def serialize_sqlalchemy_obj(obj) -> dict:
"""
序列化fetchall()后的sqlalchemy对象
https://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/
:param obj:
:return:
"""
if isinstance(obj, list):
# 转换fetchall()的结果集
return json.loads(json.dumps([dict(r) for r in obj], default=_alchemy_encoder))
else:
# 转换fetchone()后的对象
return json.loads(json.dumps(dict(obj), default=_alchemy_encoder))
对于上面的 site_info查询结果集，可以直接使用 serialize_sqlalchemy_obj方法转换成
{查询字段: 查询值} 这样的dict类型。
比起网上搜的一大堆同类文章，我这个方案，感觉优雅简单多了, 但是我这个也有缺陷，序列化了多次。