在Flask中声明一个SQLAlchemy结果集
我试图在Flask / Python中对一个SQLAlchemy结果集进行Jsonify。
Flask邮件列表build议使用以下方法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是否有一些神奇的function让邮件列表文章更容易。
编辑：澄清，这是我的模型看起来像
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
看来你实际上还没有执行你的查询。 尝试以下方法
return jsonify(json_list = qryresult.all())
[编辑] ：jsonify问题是，通常对象不能自动jsonified。 即使Python的date时间失败;）
我通常做的是为需要序列化的类添加一个额外的属性（如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 serializeable 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 serializeable 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()])
希望这可以帮助 ;）
我有同样的需要，序列化成json。 看看这个问题 。 它显示如何以编程方式发现列。 所以，我从那里创build了下面的代码。 它适用于我，我将在我的networking应用程序中使用它。 快乐的编码！
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__)
这是我的方法：
模型：
class AutoSerialize(object): 'Mixin for retrieving public fields of model in json-compatible format' __public__ = None def get_public(self, exclude=(), extra=()): "Returns model's PUBLIC data for jsonify" data = {} keys = self._sa_instance_state.attrs.items() public = self.__public__ + extra if self.__public__ else extra for k, field in keys: if public and k not in public: continue if k in exclude: continue value = self._serialize(field.value) if value: data[k] = value return data @classmethod def _serialize(cls, value, follow_fk=False): if type(value) in (datetime, date): ret = value.isoformat() elif hasattr(value, '__iter__'): ret = [] for v in value: ret.append(cls._serialize(v)) elif AutoSerialize in value.__class__.__bases__: ret = value.get_public() else: ret = value return ret class User(db.Model, AutoSerialize): __tablename__ = 'users' __public__ = ('id', 'name', 'email') id = db.Column(db.Integer, primary_key=True) name = db.Column(db.Unicode(50)) email = db.Column(db.String(120), unique=True) passhash = db.Column(db.String(100)) ...
视图：
from flask import jsonfy @mod.route('/<int:id>/', methods=['GET']) def get_user_by_id(id): u = User.query.get(id) return jsonify(u.get_public())
我不确定这一点：
self._sa_instance_state.attrs.items()
但它的作品。 我没有足够的时间让它更优雅，也许有人会build议更好的方式来获得SA领域
这对我来说通常是足够的：
我创build了一个序列化混合，我用我的模型。 序列化函数基本上获取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))
如果你正在使用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()]}
你需要明确地列出你正在返回的是什么types，我更喜欢api。 序列化很容易处理（不需要jsonify ），date也不是问题。 请注意， uri字段的内容是根据topic端点和标识自动生成的。
我一直在这个问题的一天看好，这是我想出了（指望我在这个方向https://stackoverflow.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称为Jsonify ，就像这样：
@app.route('/posts') def posts(): posts = Post.query.limit(PER_PAGE).all() return SWJsonify({'posts':posts })
似乎工作得很好。 即使在关系上。 我没有得到太多，所以YMMV，但到目前为止，我觉得这是相当“正确的”。
build议欢迎。
对于一个扁平的查询（不join），你可以做到这一点
@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)
好吧，我已经在这个工作了几个小时了，而且我已经开发了我认为是最好的pythonic解决scheme了。 下面的代码片断是python3，但是如果需要的话，不应该让backport变得非常痛苦。
我们要做的第一件事是从一个mixin开始，让你的db模型有点象dict s：
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()
现在我们要定义我们的模型，inheritancemixin：
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来让我们rest一下：
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 s，并像下面这样扩展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()))
我正在寻找类似于ActiveRecord to_json中使用的Rails方法，并在对其他build议不满意之后使用此Mixin实现了类似的function。 它处理嵌套模型，并包含或排除顶层或嵌套模型的属性。
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)
我正在处理名为jobDict的RowProxy对象列表的SQL查询defaultdict。花了我一些时间来弄清楚对象是什么types。
这是一个非常简单的快速的方法来解决一些干净的jsonEncoding只是通过types行的列表，并最初定义与列表的值。
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)
以下是在每个类上添加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)
我只是想添加我的方法来做到这一点。
只需定义一个custome json编码器来对您的db模型进行serilize。
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
然后在你的视图function
parents = Parent.query.all() dat = json.dumps({"data": parents}, cls=ParentEncoder) resp = Response(response=dat, status=200, mimetype="application/json") return (resp)
尽pipe父母有关系，但它工作得很好
这是我的答案，如果你使用的声明基地（在一些已发布的答案的帮助）：
# 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, eg 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)
这是很多次，有很多有效的答案，但下面的代码块似乎工作：
my_object = SqlAlchemyModel() my_serializable_obj = my_object.__dict__ del my_serializable_obj["_sa_instance_state"] print(jsonify(my_serializable_object))
我知道这不是一个完美的解决scheme，也不像其他人那样优雅，但是对于那些想快速修复的人来说，他们可能会尝试这个方法。
