jsonify Flask中的SQLAlchemy结果集[重复]

Posted 2023-02-15

【中文标题】jsonify Flask中的SQLAlchemy结果集[重复]

【英文标题】：jsonify a SQLAlchemy result set in Flask [duplicate]

【发布时间】：2011-10-29 11:40:53

【问题描述】：

我正在尝试在 Flask/Python 中对 SQLAlchemy 结果集进行 jsonify。

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

我在这里忽略了什么？

我发现了这个问题：How to serialize SqlAlchemy result to 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

【问题讨论】：

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

【参考方案1】：

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

return jsonify(json_list = qryresult.all())

[编辑]：jsonify 的问题是，通常对象不能被自动 json 化。甚至 Python 的日期时间也失败了 ;)

我过去所做的是向需要序列化的类添加一个额外的属性（如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 之类的库。

【讨论】：

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

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

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

@plaes 您将如何处理具有相互引用的多对多关系的两个模型？例如，User 模型具有 comments，Comment 模型具有 User 附加到它。如果你对其中任何一个调用 serialize，你会得到一个递归错误。

我并没有真正处理它。而这个答案只是一个快速的解决方案......

【参考方案2】：

这对我来说通常就足够了：

我创建了一个序列化 mixin，用于我的模型。序列化函数基本上获取 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))

【讨论】：

如果你为 get_users 使用 jsonify，你的语法需要是：return jsonify(users = User.serialize_list(users))

我认为这个答案比其他答案要好，谢谢。 inspect() 是杀手。

这对我有用。

快速简单，非常感谢！

这是一个不错的解决方案。但它不适用于需要序列化的嵌套属性。

【参考方案3】：

我也有同样的需要，序列化成 json。看看this question。它展示了如何以编程方式发现列。因此，我创建了下面的代码。它对我有用，我将在我的网络应用程序中使用它。编码愉快！

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__)

【讨论】：

这看起来很适合我当前的项目，但我使用的是非声明性模型。因此，即使在映射发生后，我似乎也无法访问班级中的__table__。关于如何将to_json 用于非声明性模型的任何想法？

我最终只是将我自己的 Table 对象添加到每个模型 (__table__ = my_table_instance) 似乎有效。

您还可以扩展声明性基类以在所有模型中自动包含json 属性。

我如何让它与datetime一起工作？我最终将sqlalchemy.sql.sqltypes.Date 添加到convert 字典中，然后将c.type 的每个实例更改为type(c.type)。

@bitcycle 在什么情况下列名会是None？

【参考方案4】：

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

pip 安装 SQLAlchemy-serializer

您可以轻松地将 mixin 添加到您的模型中，而不仅仅是调用 .to_dict() 方法在它的实例上

您也可以在 SerializerMixin 的基础上编写自己的 mixin

【讨论】：

有趣的解决方案。我必须在 elif hasattr(value, 'iter'): 在 python3 之前添加 elif isinstance(value, str): ret = value 以避免无限递归

我还取出了 if 值：签入 get_public 因为它正在丢弃整数 = 0

【参考方案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)

【讨论】：

为我工作，因为我使用了 postgresql 并在其中有一个我正在查询的表。非常感谢！

TypeError: InstanceState 类型的对象不是 JSON 可序列化的不是解决方案

【参考方案6】：

好的，我已经为此工作了几个小时，并且我开发了我认为是最 Pythonic 的解决方案。以下代码 sn-ps 是 python3，但如果您需要，向后移植应该不会太痛苦。

我们要做的第一件事是从一个 mixin 开始，让你的数据库模型表现得有点像 dicts：

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()

现在我们要定义我们的模型，继承 mixin：

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 输出包含实际未存储在数据库中的字段），那也很容易。只需将计算字段定义为@propertys，然后像这样扩展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']

现在jsonify很简单：

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

【讨论】：

我认为你的回答有点像我所做的。

很好的答案，因为它适用于原始 flask.jsonify()

【参考方案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 自动生成的。

【参考方案8】：

如果您使用的是声明式基础，这是我的答案（在一些已发布的答案的帮助下）：

# 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)

【参考方案9】：

Flask-Restful0.3.6the Request Parsing推荐棉花糖

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"

核心功能包含

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

【参考方案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)

【参考方案11】：

我在一天的大部分时间里一直在研究这个问题，这就是我想出的（感谢https://***.com/a/5249214/196358 为我指明了这个方向）。

（注意：我使用的是flask-sqlalchemy，所以我的模型声明格式与直接的sqlalchemy有点不同）。

在我的models.py 文件中：

import json

class Serializer(object):
  __public__ = None
  "Must be implemented by implementors"

  def to_serializable_dict(self):
    dict = 
    for public_key in self.__public__:
      value = getattr(self, public_key)
      if value:
        dict[public_key] = value
    return dict

class SWEncoder(json.JSONEncoder):
  def default(self, obj):
    if isinstance(obj, Serializer):
      return obj.to_serializable_dict()
    if isinstance(obj, (datetime)):
      return obj.isoformat()
    return json.JSONEncoder.default(self, obj)


def SWJsonify(*args, **kwargs):
  return current_app.response_class(json.dumps(dict(*args, **kwargs), cls=SWEncoder, indent=None if request.is_xhr else 2), mimetype='application/json')
  # stolen from https://github.com/mitsuhiko/flask/blob/master/flask/helpers.py

我所有的模型对象都是这样的：

class User(db.Model, Serializer):
  __public__ = ['id','username']
  ... field definitions ...

在我看来，无论我会调用 Jsonify，我都会调用 SWJsonify，就像这样：

@app.route('/posts')
def posts():
  posts = Post.query.limit(PER_PAGE).all()
  return SWJsonify('posts':posts )

似乎工作得很好。甚至在人际关系上。我还没有走多远，所以 YMMV，但到目前为止，它对我来说感觉很“正确”。

欢迎提出建议。

【参考方案12】：

我一直在寻找类似于 ActiveRecord to_json 中使用的 rails 方法的东西，并在对其他建议不满意后使用这个 Mixin 实现了类似的东西。它处理嵌套模型，包括或排除***或嵌套模型的属性。

class Serializer(object):

    def serialize(self, include=, exclude=[], only=[]):
        serialized = 
        for key in inspect(self).attrs.keys():
            to_be_serialized = True
            value = getattr(self, key)
            if key in exclude or (only and key not in only):
                to_be_serialized = False
            elif isinstance(value, BaseQuery):
                to_be_serialized = False
                if key in include:
                    to_be_serialized = True
                    nested_params = include.get(key, )
                    value = [i.serialize(**nested_params) for i in value]

            if to_be_serialized:
                serialized[key] = value

        return serialized

然后，为了让 BaseQuery 可序列化，我扩展了 BaseQuery

class SerializableBaseQuery(BaseQuery):

    def serialize(self, include=, exclude=[], only=[]):
        return [m.serialize(include, exclude, only) for m in self]

适用于以下型号

class ContactInfo(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    full_name = db.Column(db.String())
    source = db.Column(db.String())
    source_id = db.Column(db.String())

    email_addresses = db.relationship('EmailAddress', backref='contact_info', lazy='dynamic')
    phone_numbers = db.relationship('PhoneNumber', backref='contact_info', lazy='dynamic')


class EmailAddress(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey('contact_info.id'))


class PhoneNumber(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    phone_number = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey('contact_info.id'))

    phone_numbers = db.relationship('Invite', backref='phone_number', lazy='dynamic')

你可以这样做

@app.route("/contact/search", methods=['GET'])
def contact_search():
    contact_name = request.args.get("name")
    matching_contacts = ContactInfo.query.filter(ContactInfo.full_name.like("%%".format(contact_name)))

    serialized_contact_info = matching_contacts.serialize(
        include=
            "phone_numbers" : 
                "exclude" : ["contact_info", "contact_info_id"]
            ,
            "email_addresses" : 
                "exclude" : ["contact_info", "contact_info_id"]
            
        
    )

    return jsonify(serialized_contact_info)

【参考方案13】：

我正在使用名为 jobDict 的 RowProxy 对象列表的 sql 查询 defaultdict 我花了一段时间才弄清楚这些对象是什么类型。

这是一种非常简单的快速解决方法，只需将行类型转换为列表并最初使用 list 值定义 dict。

    jobDict = defaultdict(list)
    def set_default(obj):
        # trickyness needed here via import to know type
        if isinstance(obj, RowProxy):
            return list(obj)
        raise TypeError


    jsonEncoded = json.dumps(jobDict, default=set_default)

【参考方案14】：

我只想添加我的方法来做到这一点。

只需定义一个自定义 json 编码器来序列化您的数据库模型。

class ParentEncoder(json.JSONEncoder):
    def default(self, obj):
        # convert object to a dict
        d = 
        if isinstance(obj, Parent):
            return "id": obj.id, "name": obj.name, 'children': list(obj.child)
        if isinstance(obj, Child):
            return "id": obj.id, "name": obj.name

        d.update(obj.__dict__)
        return d

然后在你的视图函数中

parents = Parent.query.all()
dat = json.dumps("data": parents, cls=ParentEncoder)
resp = Response(response=dat, status=200, mimetype="application/json")
return (resp)

虽然父母有关系，但效果很好

【参考方案15】：

已经很多次了，有很多有效的答案，但下面的代码块似乎有效：

my_object = SqlAlchemyModel()
my_serializable_obj = my_object.__dict__
del my_serializable_obj["_sa_instance_state"]
print(jsonify(my_serializable_object))

我知道这不是一个完美的解决方案，也不像其他解决方案那样优雅，但是对于那些想要快速修复的人来说，他们可能会尝试这个。