sqlalchemy 查询结果集最简单序列化serialize 处理时间和小数

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类型。

比起网上搜的一大堆同类文章,我这个方案,感觉优雅简单多了, 但是我这个也有缺陷,序列化了多次。


文章作者: 王小右
版权声明: 咳咳想白嫖文章?本文章著作权归作者所有,任何形式的转载都请注明出处。 https://www.charmcode.cn !
  目录