在Flask中将SQLAlchemy的结果集转换为JSON格式

Question

我正在尝试在Flask/Python中将SQLAlchemy结果集转换为JSON。

Flask邮件列表建议使用以下方法 http://librelist.com/browser//flask/2011/2/16/jsonify-sqlalchemy-pagination-collection-result/#04a0754b63387f87e59dda564bde426e ：

return jsonify(json_list = qryresult)

然而我收到了以下错误信息：

TypeError: <flaskext.sqlalchemy.BaseQuery object at 0x102c2df90> 
is not JSON serializable

我在这里忽略了什么吗？

我找到了这个问题：如何将SqlAlchemy结果序列化为JSON？，它看起来非常相似，但是我不知道Flask是否有一些魔法可以使它更容易，就像邮件列表帖子所建议的那样。

编辑：为了澄清，这就是我的模型长什么样子。

class Rating(db.Model):

    __tablename__ = 'rating'

    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    overall = db.Column(db.Integer)
    shipping = db.Column(db.Integer)
    cost = db.Column(db.Integer)
    honesty = db.Column(db.Integer)
    communication = db.Column(db.Integer)
    name = db.Column(db.String())
    ipaddr = db.Column(db.String())
    date = db.Column(db.String())

    def __init__(self, fullurl, url, comments, overall, shipping, cost, honesty, communication, name, ipaddr, date):
        self.fullurl = fullurl
        self.url = url
        self.comments = comments
        self.overall = overall
        self.shipping = shipping
        self.cost = cost
        self.honesty = honesty
        self.communication = communication
        self.name = name
        self.ipaddr = ipaddr
        self.date = date

- mal-wan

---

jsonify(list(map(lambda x: x.to_dict(), qryresult))) - Aabesh Karmacharya

Answer 1

看起来你实际上还没有执行你的查询。请尝试以下操作：

return jsonify(json_list = qryresult.all())

[编辑]：jsonify 的问题在于通常不能自动将对象转换为 JSON 格式。即使 Python 的 datetime 也无法转换 ;)

我过去所做的是，在需要被序列化的类中添加额外的属性（例如 serialize）。

def dump_datetime(value):
    """Deserialize datetime object into string form for JSON processing."""
    if value is None:
        return None
    return [value.strftime("%Y-%m-%d"), value.strftime("%H:%M:%S")]

class Foo(db.Model):
    # ... SQLAlchemy defs here..
    def __init__(self, ...):
       # self.foo = ...
       pass

    @property
    def serialize(self):
       """Return object data in easily serializable format"""
       return {
           'id'         : self.id,
           'modified_at': dump_datetime(self.modified_at),
           # This is an example how to deal with Many2Many relations
           'many2many'  : self.serialize_many2many
       }
    @property
    def serialize_many2many(self):
       """
       Return object's relations in easily serializable format.
       NB! Calls many2many's serialize property.
       """
       return [ item.serialize for item in self.many2many]

现在，我只需要这样处理视图：

return jsonify(json_list=[i.serialize for i in qryresult.all()])

[2019年修订]: 如果您有更复杂的对象或循环引用，请使用像marshmallow这样的库。

- plaes

---

嗯，这改变了错误，现在我得到的错误是关于SQLAlchemy对象的引用，如：myapp.models.Rating object at 0x102f25c10> is not JSON serializable。有什么线索吗？该对象只包含字符串和整数。 - mal-wan

想知道为什么您选择将serialize作为属性而不是函数？ - Localist

@Mohamed 7年前这是有道理的。我创建了serialize作为一个属性，因为我不需要向它推送任何参数。当然，as_json可能是更好的名称。 - plaes

@plaes 如果两个模型之间存在多对多关系并且相互引用，你会如何处理？例如，一个User模型有comments，而Comment模型则附加了一个User。如果你在任何一个模型上调用序列化操作，都会出现递归错误。 - Darius Mandres

我并不真正处理它。而且这个答案只是一个快速解决方案... - plaes

Answer 2

以下是通常对我来说足够的做法：

我创建了一个序列化混合类，然后将其与我的模型一起使用。序列化函数基本上会获取SQLAlchemy检查器公开的所有属性，并将其放入字典中。

from sqlalchemy.inspection import inspect

class Serializer(object):

    def serialize(self):
        return {c: getattr(self, c) for c in inspect(self).attrs.keys()}

    @staticmethod
    def serialize_list(l):
        return [m.serialize() for m in l]

现在需要做的就是使用Serializer mixin类扩展SQLAlchemy模型。

如果有您不希望公开或需要特殊格式的字段，只需在模型子类中重写serialize()函数即可。

class User(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String)
    password = db.Column(db.String)

    # ...

    def serialize(self):
        d = Serializer.serialize(self)
        del d['password']
        return d

在你的控制器中，你只需要调用结果上的serialize()函数 (或者如果查询结果是一个列表，则调用serialize_list(l))：

def get_user(id):
    user = User.query.get(id)
    return json.dumps(user.serialize())

def get_users():
    users = User.query.all()
    return json.dumps(User.serialize_list(users))

- Carl Ekerot

---

如果您在使用 jsonify 函数来获取用户信息时，需要写成以下语法形式： return jsonify(users=User.serialize_list(users))。这将使用户信息序列化后以 JSON 的格式返回。 - idbill

我认为这个答案比其他答案好多了，谢谢。inspect() 真是太棒了。 - robru

这是一个不错的解决方案。但对于需要序列化的嵌套属性，它将无法正常工作。 - Faizi

这太棒了，谢谢！ - antonmills

我不得不自定义序列化方法以跳过我的外键，否则它会报错“myForeignObject不可序列化”。（这些属性是使用db.relationship(...)设置的） - Alex

Answer 3

我有同样的需求，要将数据序列化为JSON格式。可以看看this question，它展示了如何以编程方式发现列。基于此，我编写了以下代码。它对我有效，并且我会在我的Web应用程序中使用它。

def to_json(inst, cls):
    """
    Jsonify the sql alchemy query result.
    """
    convert = dict()
    # add your coversions for things like datetime's 
    # and what-not that aren't serializable.
    d = dict()
    for c in cls.__table__.columns:
        v = getattr(inst, c.name)
        if c.type in convert.keys() and v is not None:
            try:
                d[c.name] = convert[c.type](v)
            except:
                d[c.name] = "Error:  Failed to covert using ", str(convert[c.type])
        elif v is None:
            d[c.name] = str()
        else:
            d[c.name] = v
    return json.dumps(d)

class Person(base):
    __tablename__ = 'person'
    id = Column(Integer, Sequence('person_id_seq'), primary_key=True)
    first_name = Column(Text)
    last_name = Column(Text)
    email = Column(Text)

    @property
    def json(self):
        return to_json(self, self.__class__)

- bitcycle

---

这看起来很适合我的当前项目，但我正在使用非声明性模型。因此，在映射发生后，即使在类中也似乎无法访问__table__。有没有想过如何为非声明性模型调整to_json的方法？ - technomalogical

最终我只是在每个模型中添加了自己的“Table”对象（__table__ = my_table_instance），这似乎起作用了。 - technomalogical

此外，您还可以扩展声明性的基类，以便自动在所有模型中包含json属性。 - edsioufi

我该如何使用datetime使其正常工作？最终，我将sqlalchemy.sql.sqltypes.Date添加到了convert字典中，然后我将每个c.type实例更改为type(c.type)。 - Matthew Moisen

@bitcycle 在哪些情况下列名将为None？ - Layla

这里有一个类似的解决方案：https://wakatime.com/blog/32-part-1-sqlalchemy-models-to-json - Alan Hamlett

Answer 4

这是我的方法： https://github.com/n0nSmoker/SQLAlchemy-serializer

pip install SQLAlchemy-serializer

您可以轻松地将mixin添加到您的模型中，然后只需在其实例上调用.to_dict()方法。

您还可以基于SerializerMixin编写自己的mixin。

- n0nSmoker

---

有趣的解决方案。在Python3中，我不得不添加以下代码才能避免无限递归：elif isinstance(value, str): ret = value这行代码要放在elif hasattr(value, '__iter__'):之前。 - Shaun

我还在get_public函数中删除了if value: check，因为它会丢弃整数= 0。 - Shaun

Answer 5

对于一个没有连接的平面查询，您可以这样做。

@app.route('/results/')
def results():
    data = Table.query.all()
    result = [d.__dict__ for d in data]
    return jsonify(result=result)

如果你只想从数据库返回特定的列，可以这样做：

@app.route('/results/')
def results():
    cols = ['id', 'url', 'shipping']
    data = Table.query.all()
    result = [{col: getattr(d, col) for col in cols} for d in data]
    return jsonify(result=result)

- reubano

---

对我来说很有效，因为我使用了PostgreSQL，并在其中有一个表格进行查询。非常感谢！ - h612

类型错误：InstanceState对象不可序列化为JSON。这并不是一个解决方案。 - fsalazar_sch

我也遇到了与@fsalazar_sch相同的错误：TypeError: Object of type InstanceState is not JSON serializable。 - Rylan Schaeffer

Answer 6

好的，我已经花了几个小时在这上面工作了，我开发出了我认为是最具Python风格的解决方案。以下代码片段是用Python3编写的，但如果您需要的话，不应该太痛苦地进行回溯。

我们要做的第一件事是使用mixin让您的数据库模型像dict一样运行：

from sqlalchemy.inspection import inspect

class ModelMixin:
    """Provide dict-like interface to db.Model subclasses."""

    def __getitem__(self, key):
        """Expose object attributes like dict values."""
        return getattr(self, key)

    def keys(self):
        """Identify what db columns we have."""
        return inspect(self).attrs.keys()

现在我们将定义我们的模型，继承该混合类：

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)
    # etc ...

那么，只需要传递 MyModel() 的实例给 dict() 函数，就能够得到一个真正的 dict 实例，这让我们在让 jsonify() 处理它时更加方便。接下来，我们需要扩展 JSONEncoder 来完成其余部分：

from flask.json import JSONEncoder
from contextlib import suppress

class MyJSONEncoder(JSONEncoder):
    def default(self, obj):
        # Optional: convert datetime objects to ISO format
        with suppress(AttributeError):
            return obj.isoformat()
        return dict(obj)

app.json_encoder = MyJSONEncoder

加分项：如果你的模型包含计算字段（也就是说，你希望在JSON输出中包含实际上并没有存储在数据库中的字段），也很容易。只需将计算字段定义为@property，并像这样扩展keys()方法：

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)

    @property
    def computed_field(self):
        return 'this value did not come from the db'

    def keys(self):
        return super().keys() + ['computed_field']

现在将其转换为JSON格式非常简单：

@app.route('/whatever', methods=['GET'])
def whatever():
    return jsonify(dict(results=MyModel.query.all()))

- robru

---

我认为你的答案有点像我所做的。 - user4985526

很好的答案，因为它可以与原始的flask.jsonify()一起使用。 - Michael Ribbons

Answer 7

如果你正在使用flask-restful，你可以使用marshal：

from flask.ext.restful import Resource, fields, marshal

topic_fields = {
    'title':   fields.String,
    'content': fields.String,
    'uri':     fields.Url('topic'),
    'creator': fields.String,
    'created': fields.DateTime(dt_format='rfc822')
}

class TopicListApi(Resource):
    def get(self):
        return {'topics': [marshal(topic, topic_fields) for topic in DbTopic.query.all()]}

你需要明确列出你正在返回的内容及其类型，对于 API，我更喜欢这种方式。序列化很容易处理（不需要使用 jsonify），日期也不是问题。请注意，uri 字段的内容是基于 topic 端点和 ID 自动生成的。

- Adversus

Answer 8

Flask-Restful 0.3.6中的请求参数解析推荐使用marshmallow。

marshmallow是一个ORM/ODM/框架无关的库，用于将复杂的数据类型（例如对象）转换为原生Python数据类型，并在它们之间进行相互转换。

下面是一个简单的marshmallow示例。

from marshmallow import Schema, fields

class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

from marshmallow import pprint

user = User(name="Monty", email="[e-mail]")
schema = UserSchema()
result = schema.dump(user)
pprint(result)
# {"name": "Monty",
#  "email": "[e-mail]",
#  "created_at": "2014-08-17T14:54:16.049594+00:00"}

核心功能包括：

声明模式
对象序列化（“转储”）
对象反序列化（“加载”）
处理对象集合
验证
指定属性名称
指定序列化/反序列化键
重构：隐式字段创建
排序输出
“只读”和“只写”字段
指定默认的序列化/反序列化值
嵌套模式
自定义字段

- Shihe Zhang

Answer 9

如果您正在使用声明性基础（借助已发布的某些答案），我将为您提供答案：

在这里是我的答案：

# in your models definition where you define and extend declarative_base()
from sqlalchemy.ext.declarative import declarative_base
...
Base = declarative_base()
Base.query = db_session.query_property()
...

# define a new class (call "Model" or whatever) with an as_dict() method defined
class Model():
    def as_dict(self):
        return { c.name: getattr(self, c.name) for c in self.__table__.columns }

# and extend both the Base and Model class in your model definition, e.g.
class Rating(Base, Model):
    ____tablename__ = 'rating'
    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    ...

# then after you query and have a resultset (rs) of ratings
rs = Rating.query.all()

# you can jsonify it with
s = json.dumps([r.as_dict() for r in rs], default=alchemyencoder)
print (s)

# or if you have a single row
r = Rating.query.first()

# you can jsonify it with
s = json.dumps(r.as_dict(), default=alchemyencoder)

# you will need this alchemyencoder where your are calling json.dumps to handle datetime and decimal format
# credit to Joonas @ http://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/
def alchemyencoder(obj):
    """JSON encoder function for SQLAlchemy special classes."""
    if isinstance(obj, datetime.date):
        return obj.isoformat()
    elif isinstance(obj, decimal.Decimal):
        return float(obj)

- VinnyQ77

Answer 10

下面是一种为每个类添加as_dict()方法的方式，以及您想要在每个单独类中拥有的任何其他方法。

我不确定这是否是所期望的方式，但它是可行的...

class Base(object):
    def as_dict(self):
        return dict((c.name,
                     getattr(self, c.name))
                     for c in self.__table__.columns)


Base = declarative_base(cls=Base)

- tahoe