python sqlalchemy JSON

SQLAlchemy转换成json格式

SQLAlchemy查询出来object转换成json格式，如果数据模型中没有使用外键的话。那么可以直接调用object中的dict方法即可将object转换成dict格式，之后json.dumps即可。

如果数据模型中使用了外键，在调用dict方法后，会有个一个”_sa_instance_state”的key，其value则是一个object，对于外键有很多个可配置对参数：

backref:在一对多或多对一之间建立双向关系
lazy:默认值是True,说明关联对象只有到真正访问的时候才会去查询数据库,比如有parent对象,只有知道访问parent.children的时候才做关联查询，这就是上面所说的在调用了dict后，对于的外键关联会生成一个关联表的查询object。

对于存在外键的查询格式化成json的需求，可以通过如下代码来实现：

import json
 
 
class ComplexEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime(‘%Y-%m-%d %H:%M:%S‘)
        elif isinstance(obj, datetime.date):
            return obj.strftime(‘%Y-%m-%d‘)
        else:
            return json.JSONEncoder.default(self, obj)
 
 
def obj_to_json(obj_list):
    out = [q.__dict__ for q in obj_list]
    for objs, instance in zip(out, obj_list):
        for obj in objs.values():
            if callable(obj):
                for name in obj.mapper.relationships.keys():
                    tmp = getattr(instance, name).__dict__
                    if "_sa_instance_state" in tmp.keys():
                        tmp.pop("_sa_instance_state")
                        tmp.pop("id")
                        objs.update(tmp)
                    objs.pop(name)
        if "_sa_instance_state" in objs.keys():
            objs.pop("_sa_instance_state")
    return out
 
 
def query_result_json(query_result):
    """
    Convert query result to json format
    """
    if isinstance(query_result, list):
        result = obj_to_json(query_result)
    elif getattr(query_result, ‘__dict__‘, ‘‘):
        result = obj_to_json([query_result])
    else:
        result = {‘result‘: query_result}
    return json.dumps(result, cls=ComplexEncoder)import json
 
 
class ComplexEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime.datetime):
            return obj.strftime(‘%Y-%m-%d %H:%M:%S‘)
        elif isinstance(obj, datetime.date):
            return obj.strftime(‘%Y-%m-%d‘)
        else:
            return json.JSONEncoder.default(self, obj)
 
 
def obj_to_json(obj_list):
    out = [q.__dict__ for q in obj_list]
    for objs, instance in zip(out, obj_list):
        for obj in objs.values():
            if callable(obj):
                for name in obj.mapper.relationships.keys():
                    tmp = getattr(instance, name).__dict__
                    if "_sa_instance_state" in tmp.keys():
                        tmp.pop("_sa_instance_state")
                        tmp.pop("id")
                        objs.update(tmp)
                    objs.pop(name)
        if "_sa_instance_state" in objs.keys():
            objs.pop("_sa_instance_state")
    return out
 
 
def query_result_json(query_result):
    """
    Convert query result to json format
    """
    if isinstance(query_result, list):
        result = obj_to_json(query_result)
    elif getattr(query_result, ‘__dict__‘, ‘‘):
        result = obj_to_json([query_result])
    else:
        result = {‘result‘: query_result}
    return json.dumps(result, cls=ComplexEncoder)