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

Question

Django有一个很不错的功能，可以把从数据库中取出的ORM模型自动转换成JSON格式。

那么，如何把SQLAlchemy查询的结果转换成JSON格式呢？

我试过用 jsonpickle.encode，但它只是把查询对象本身编码了。我也试过用 json.dumps(items)，但返回的结果是

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

难道把SQLAlchemy的ORM对象转换成JSON或XML真的这么难吗？难道没有默认的序列化工具可以用吗？现在把ORM查询结果转换成JSON是个很常见的任务。

我需要的只是把SQLAlchemy查询结果以JSON或XML的形式返回。

把SQLAlchemy对象的查询结果转换成JSON/XML格式是为了在JavaScript的数据表格（JQGrid http://www.trirand.com/blog/）中使用。

Answer 1

Python 3.7+ 和 Flask 1.1+ 可以使用内置的 数据类 包

from dataclasses import dataclass
from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy


app = Flask(__name__)
db = SQLAlchemy(app)


@dataclass
class User(db.Model):
  id: int
  email: str

  id = db.Column(db.Integer, primary_key=True, auto_increment=True)
  email = db.Column(db.String(200), unique=True)


@app.route('/users/')
def users():
  users = User.query.all()
  return jsonify(users)  


if __name__ == "__main__":
  users = User(email="user1@gmail.com"), User(email="user2@gmail.com")
  db.create_all()
  db.session.add_all(users)
  db.session.commit()
  app.run()

现在，/users/ 这个路径会返回一个用户列表。

[
  {"email": "user1@gmail.com", "id": 1},
  {"email": "user2@gmail.com", "id": 2}
]

自动序列化相关模型

@dataclass
class Account(db.Model):
  id: int
  users: User

  id = db.Column(db.Integer)
  users = db.relationship(User)  # User model would need a db.ForeignKey field

使用 jsonify(account) 得到的响应会是这样的。

{  
   "id":1,
   "users":[  
      {  
         "email":"user1@gmail.com",
         "id":1
      },
      {  
         "email":"user2@gmail.com",
         "id":2
      }
   ]
}

覆盖默认的 JSON 编码器

from flask.json import JSONEncoder


class CustomJSONEncoder(JSONEncoder):
  "Add support for serializing timedeltas"

  def default(o):
    if type(o) == datetime.timedelta:
      return str(o)
    if type(o) == datetime.datetime:
      return o.isoformat()
    return super().default(o)

app.json_encoder = CustomJSONEncoder      

Answer 2

你可以把你的对象输出成一个字典：

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

然后你可以使用 User.as_dict() 来把你的对象转换成可以存储的格式。

具体的解释可以参考这个链接：将sqlalchemy行对象转换为python字典

Answer 3

一种简单的实现方式

你可以使用类似这样的代码：

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)

这个方法会忽略那些无法编码的字段（会把它们设置为'None'）。

它不会自动展开关系（因为这样可能会导致自我引用，造成无限循环）。

一种递归的、非循环的实现方式

不过，如果你想要无限循环的话，可以使用：

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)

这样会编码所有的子对象，以及它们的子对象，依此类推……基本上可能会编码你整个数据库。当它遇到之前编码过的内容时，会把它编码为'None'。

一种递归的、可能循环的、选择性实现方式

还有另一种选择，可能更好，就是可以指定你想要展开的字段：

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字段。