如何将SqlAlchemy结果序列化为JSON？

Question

Django对从DB返回到JSON格式的ORM模型有一些很好的自动序列化。

如何将SQLAlchemy查询结果序列化为JSON格式？

我试过jsonpickle.encode，但它对查询对象本身进行编码。 我试过json.dumps(items)但它返回了

TypeError: <Product('3', 'some name', 'some desc')> is not JSON serializable

将SQLAlchemy ORM对象序列化为JSON/XML真的那么难吗？没有默认的序列化程序吗？序列化ORM查询结果是当前非常常见的任务。

我需要的只是返回SQLAlchemy查询结果的JSON或XML数据表示。

javascript datagird（JQGridhttp://www.trirand.com/blog/）中需要使用JSON/XML格式的SQLAlchemy对象查询结果

Answer 1

简单的实现

你可以用这样的东西：

from sqlalchemy.ext.declarative import DeclarativeMeta

class AlchemyEncoder(json.JSONEncoder):

    def default(self, obj):
        if isinstance(obj.__class__, DeclarativeMeta):
            # an SQLAlchemy class
            fields = {}
            for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                data = obj.__getattribute__(field)
                try:
                    json.dumps(data) # this will fail on non-encodable values, like other classes
                    fields[field] = data
                except TypeError:
                    fields[field] = None
            # a json-encodable dict
            return fields

        return json.JSONEncoder.default(self, obj)

然后使用以下命令转换为JSON：

c = YourAlchemyClass()
print json.dumps(c, cls=AlchemyEncoder)

它将忽略不可编码的字段（将它们设置为“无”）。

它不会自动扩展关系（因为这可能导致自引用，并永远循环）。

递归的、非循环的实现

但是，如果您希望永远循环，您可以使用：

from sqlalchemy.ext.declarative import DeclarativeMeta

def new_alchemy_encoder():
    _visited_objs = []

    class AlchemyEncoder(json.JSONEncoder):
        def default(self, obj):
            if isinstance(obj.__class__, DeclarativeMeta):
                # don't re-visit self
                if obj in _visited_objs:
                    return None
                _visited_objs.append(obj)

                # an SQLAlchemy class
                fields = {}
                for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                    fields[field] = obj.__getattribute__(field)
                # a json-encodable dict
                return fields

            return json.JSONEncoder.default(self, obj)

    return AlchemyEncoder

然后使用以下方法对对象进行编码：

print json.dumps(e, cls=new_alchemy_encoder(), check_circular=False)

这将编码所有的孩子，所有的孩子，所有的孩子。。。基本上可以对整个数据库进行编码。当它到达之前编码的东西时，它会将其编码为“无”。

递归的，可能是循环的，选择性的实现

另一个可能更好的选择是能够指定要展开的字段：

def new_alchemy_encoder(revisit_self = False, fields_to_expand = []):
    _visited_objs = []

    class AlchemyEncoder(json.JSONEncoder):
        def default(self, obj):
            if isinstance(obj.__class__, DeclarativeMeta):
                # don't re-visit self
                if revisit_self:
                    if obj in _visited_objs:
                        return None
                    _visited_objs.append(obj)

                # go through each field in this SQLalchemy class
                fields = {}
                for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
                    val = obj.__getattribute__(field)

                    # is this field another SQLalchemy object, or a list of SQLalchemy objects?
                    if isinstance(val.__class__, DeclarativeMeta) or (isinstance(val, list) and len(val) > 0 and isinstance(val[0].__class__, DeclarativeMeta)):
                        # unless we're expanding this field, stop here
                        if field not in fields_to_expand:
                            # not expanding this field: set it to None and continue
                            fields[field] = None
                            continue

                    fields[field] = val
                # a json-encodable dict
                return fields

            return json.JSONEncoder.default(self, obj)

    return AlchemyEncoder

您现在可以用以下方法调用它：

print json.dumps(e, cls=new_alchemy_encoder(False, ['parents']), check_circular=False)

例如，只扩展名为“parents”的SQLAlchemy字段。

Answer 2

您可以将RowProxy转换为如下所示的dict：

 d = dict(row.items())

然后将其序列化为JSON（必须为诸如datetime值之类的内容指定编码器） 如果你只想要一条记录（而不是一个完整的相关记录层次结构），这并不难。

json.dumps([(dict(row.items())) for row in rs])

Answer 3

您可以将对象作为字典输出：

class User:
   def as_dict(self):
       return {c.name: getattr(self, c.name) for c in self.__table__.columns}

然后使用User.as_dict()序列化对象。

如Convert sqlalchemy row object to python dict所述