在 Flask 中 jsonify SQLAlchemy 结果集

小开

最佳答案

似乎您实际上还没有执行您的查询:

return jsonify(json_list = qryresult.all())

[ Edit ] : jsonify 的问题是，通常对象不能自动 jsonify，甚至 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] : 如果您有更复杂的对象或循环引用，请使用类似棉花糖的库。

小开

我也有同样的需求，想要连载到 Json 身上。看看这个问题。它演示如何以编程方式发现列。因此，从那里我创建了下面的代码。它对我很有用，我会在我的网络应用程序中使用它。编程愉快！


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

小开

我花了大半天的时间来研究这个问题，以下是我所得到的结果(感谢 https://stackoverflow.com/a/5249214/196358为我指出了这个方向)。

(注意: 我使用的是 flask-sql伟金术，因此我的模型声明格式与直接的 sql伟金术稍有不同)。

在我的 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，但到目前为止，它感觉相当“正确的”给我。

欢迎提出建议。

小开

我的方法是: Https://github.com/n0nsmoker/sqlalchemy-serializer

pip install SQLAlchemy-serializer

您可以很容易地向模型中添加 Mixin，然后只需调用实例上的 .to_dict()方法。

你也可以在 SerializerMixin的基础上编写你自己的混音。

小开

对于平面查询(无连接) ，可以这样做

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

小开

我正在处理一个名为 jobDect 的 RowProxy 对象列表的 sql 查询默认值我花了一段时间才弄清楚这些物体是什么类型的。

这是一种非常简单的快速方法，可以通过将行类型转换为列表并最初使用 list 值定义 dict 来解决一些干净的 jsonEncoding 问题。

    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)

小开

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

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

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

小开

这些对我来说已经足够了:

我创建了一个序列化混合器，我在其中使用我的模型。序列化函数基本上获取 SQLAlchemy 检查器公开的任何属性，并将其放入一个 dict 中。

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混合类扩展 SQLAlchemy 模型。

如果有些字段您不希望公开，或者需要特殊的格式，只需覆盖 model 子类中的 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))

小开

这里有一种方法可以在每个类上添加 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)

小开

好吧，我已经研究了几个小时了，我已经开发出了我认为是迄今为止最简洁的解决方案。下面的代码片段是 python3，但是如果需要的话，应该不会对后端口造成太大的影响。

我们要做的第一件事情就是从混合模式开始，让你的数据库模型有点像 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()

现在我们要定义我们的模型，继承混合:

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']

现在，jsonify 变得微不足道了:

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

小开

如果你使用的是 flask-restful，你可以使用法警:

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 自动生成的。

小开

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

只需定义一个自定义的 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)

尽管父母之间有关系，但是这种方法很有效

小开

我正在寻找 ActiveRecord to _ json 中使用的 Rail 方法，并在对其他建议不满意之后使用 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)

小开

如果你使用声明式基础，以下是我的答案(参考一些已经发布的答案) :

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

小开

推荐棉花糖

棉花糖是一个 ORM/ODM/框架无关的库，用于转换复杂数据类型(如对象)与原生 Python 之间的往来数据类型。

下面是一个简单的棉花糖示例。

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="monty@python.org")
schema = UserSchema()
result = schema.dump(user)
pprint(result)
# {"name": "Monty",
#  "email": "monty@python.org",
#  "created_at": "2014-08-17T14:54:16.049594+00:00"}

核心特性包括

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