如何将SqlAlchemy结果序列化为JSON？
Django有一些很好的从DB到JSON格式返回的ORM模型的自动序列化。
如何将SQLAlchemy查询结果序列化为JSON格式？
我试过jsonpickle.encode但它编码查询对象本身。 我试过json.dumps(items)但它返回
TypeError: <Product('3', 'some name', 'some desc')> is not JSON serializable
将SQLAlchemy ORM对象序列化为JSON / XML真的很难吗？ 没有任何默认序列化程序？ 现在序列化ORM查询结果是很常见的任务。
我需要的只是返回SQLAlchemy查询结果的JSON或XML数据表示。
JSON格式的SQLAlchemy对象查询结果需要在javascript datagird中使用（JQGrid http://www.trirand.com/blog/ ）
平坦的实现
你可以使用这样的东西：
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)
它会忽略不可编码的字段（将它们设置为“无”）。
它不会自动扩展关系（因为这可能导致自引用，并永远循环）。
recursion的，非循环的实现
但是，如果你宁愿永远循环，你可以使用：
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)
这将编码所有的孩子，和他们所有的孩子，以及他们所有的孩子…基本上可能编码你的整个数据库。 当它达到之前编码的东西时，它会将其编码为“无”。
recursion的，可能是循环的select性实现
另一种可能更好的方法是能够指定要扩展的字段：
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)
例如，仅扩展名为“父母”的SQLAlchemy字段。
你可以输出你的对象作为字典：
class User: def as_dict(self): return {c.name: getattr(self, c.name) for c in self.__table__.columns}
然后你使用User.as_dict（）来序列化你的对象。
正如将sqlalchemy行对象转换为python字典所解释的那样
你可以把RowProxy转换成这样的字典：
d = dict(row.items())
然后将其序列化为JSON（您将不得不指定编码器来处理datetime值）如果您只需要一条logging（而不是完整的相关logging层次结构），那就不难了。
json.dumps([(dict(row.items())) for row in rs])
我build议使用最近浮出水面的图书馆棉花糖 。 它允许您创build序列化程序来表示支持关系和嵌套对象的模型实例。
看看这个SQLAlchemy的例子 。
Flask-JsonTools包为您的模型实现了JsonSerializableBase基类。
用法：
from sqlalchemy.ext.declarative import declarative_base from flask.ext.jsontools import JsonSerializableBase Base = declarative_base(cls=(JsonSerializableBase,)) class User(Base): #...
现在User模型是神奇的序列化。
如果你的框架不是Flask，那么你可以抓住代码
出于安全原因，您不应该返回所有模型的字段。 我更喜欢有select地select它们。
Flask的json编码现在支持UUID，date时间和关系（并为flask_sqlalchemy db.Model类添加了query和query_class ）。 我已经更新了编码器，如下所示：
应用程序/ json_encoder.py
from sqlalchemy.ext.declarative import DeclarativeMeta from flask import json class AlchemyEncoder(json.JSONEncoder): def default(self, o): if isinstance(o.__class__, DeclarativeMeta): data = {} fields = o.__json__() if hasattr(o, '__json__') else dir(o) for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class']]: value = o.__getattribute__(field) try: json.dumps(value) data[field] = value except TypeError: data[field] = None return data return json.JSONEncoder.default(self, o)
app/__init__.py
# json encoding from app.json_encoder import AlchemyEncoder app.json_encoder = AlchemyEncoder
有了这个，我可以select添加一个__json__属性，返回我希望编码的字段列表：
app/models.py
class Queue(db.Model): id = db.Column(db.Integer, primary_key=True) song_id = db.Column(db.Integer, db.ForeignKey('song.id'), unique=True, nullable=False) song = db.relationship('Song', lazy='joined') type = db.Column(db.String(20), server_default=u'audio/mpeg') src = db.Column(db.String(255), nullable=False) created_at = db.Column(db.DateTime, server_default=db.func.now()) updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now()) def __init__(self, song): self.song = song self.src = song.full_path def __json__(self): return ['song', 'src', 'type', 'created_at']
我添加@jsonapi到我的视图，返回结果列表，然后我的输出如下：
[ { "created_at": "Thu, 23 Jul 2015 11:36:53 GMT", "song": { "full_path": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3", "id": 2, "path_name": "Audioslave/Audioslave [2002]/1 Cochise.mp3" }, "src": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3", "type": "audio/mpeg" } ]
这并不那么直截了当。 我写了一些代码来做到这一点。 我还在研究它，它使用MochiKit框架。 它基本上使用代理和注册的JSON转换器来翻译Python和Javascript之间的复合对象。
数据库对象的浏览器端是db.js它需要proxy.js中的基本Python代理源。
在Python方面有基本的代理模块 。 然后终于在webserver.py中的SqlAlchemy对象编码器。 它还取决于models.py文件中的元数据提取器。
自定义序列化和反序列化。
“from_json” （类方法）基于json数据构build一个Model对象。
“deserialize”只能在实例上调用，并将json中的所有数据合并到Model实例中。
“序列化” – recursion序列化
需要__write_only__属性来定义只写属性（例如“password_hash”）。
class Serializable(object): __exclude__ = ('id',) __include__ = () __write_only__ = () @classmethod def from_json(cls, json, selfObj=None): if selfObj is None: self = cls() else: self = selfObj exclude = (cls.__exclude__ or ()) + Serializable.__exclude__ include = cls.__include__ or () if json: for prop, value in json.iteritems(): # ignore all non user data, eg only if (not (prop in exclude) | (prop in include)) and isinstance( getattr(cls, prop, None), QueryableAttribute): setattr(self, prop, value) return self def deserialize(self, json): if not json: return None return self.__class__.from_json(json, selfObj=self) @classmethod def serialize_list(cls, object_list=[]): output = [] for li in object_list: if isinstance(li, Serializable): output.append(li.serialize()) else: output.append(li) return output def serialize(self, **kwargs): # init write only props if len(getattr(self.__class__, '__write_only__', ())) == 0: self.__class__.__write_only__ = () dictionary = {} expand = kwargs.get('expand', ()) or () prop = 'props' if expand: # expand all the fields for key in expand: getattr(self, key) iterable = self.__dict__.items() is_custom_property_set = False # include only properties passed as parameter if (prop in kwargs) and (kwargs.get(prop, None) is not None): is_custom_property_set = True iterable = kwargs.get(prop, None) # loop trough all accessible properties for key in iterable: accessor = key if isinstance(key, tuple): accessor = key[0] if not (accessor in self.__class__.__write_only__) and not accessor.startswith('_'): # force select from db to be able get relationships if is_custom_property_set: getattr(self, accessor, None) if isinstance(self.__dict__.get(accessor), list): dictionary[accessor] = self.__class__.serialize_list(object_list=self.__dict__.get(accessor)) # check if those properties are read only elif isinstance(self.__dict__.get(accessor), Serializable): dictionary[accessor] = self.__dict__.get(accessor).serialize() else: dictionary[accessor] = self.__dict__.get(accessor) return dictionary
使用SQLAlchemy中的内置序列化程序 ：
from sqlalchemy.ext.serializer import loads, dumps obj = MyAlchemyObject() # serialize object serialized_obj = dumps(obj) # deserialize object obj = loads(serialized_obj)
如果要在会话之间传输对象，请记住使用session.expunge(obj)将对象从当前会话中分离出来。 再次附加，只要做session.add(obj) 。
这里有一个解决scheme，可以让你select你想要包含在你的输出中的关系。 注意：这是一个完整的重写，把一个字典/ str作为一个arg而不是一个列表。 修复了一些东西..
def deep_dict(self, relations={}): """Output a dict of an SA object recursing as deep as you want. Takes one argument, relations which is a dictionary of relations we'd like to pull out. The relations dict items can be a single relation name or deeper relation names connected by sub dicts Example: Say we have a Person object with a family relationship person.deep_dict(relations={'family':None}) Say the family object has homes as a relation then we can do person.deep_dict(relations={'family':{'homes':None}}) OR person.deep_dict(relations={'family':'homes'}) Say homes has a relation like rooms you can do person.deep_dict(relations={'family':{'homes':'rooms'}}) and so on... """ mydict = dict((c, str(a)) for c, a in self.__dict__.items() if c != '_sa_instance_state') if not relations: # just return ourselves return mydict # otherwise we need to go deeper if not isinstance(relations, dict) and not isinstance(relations, str): raise Exception("relations should be a dict, it is of type {}".format(type(relations))) # got here so check and handle if we were passed a dict if isinstance(relations, dict): # we were passed deeper info for left, right in relations.items(): myrel = getattr(self, left) if isinstance(myrel, list): mydict[left] = [rel.deep_dict(relations=right) for rel in myrel] else: mydict[left] = myrel.deep_dict(relations=right) # if we get here check and handle if we were passed a string elif isinstance(relations, str): # passed a single item myrel = getattr(self, relations) left = relations if isinstance(myrel, list): mydict[left] = [rel.deep_dict(relations=None) for rel in myrel] else: mydict[left] = myrel.deep_dict(relations=None) return mydict
所以例如使用人/家庭/家庭/房间…把它变成json，你需要的就是
json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))
def alc2json(row): return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])
我以为我会用这个打高尔夫球。
仅供参考：我正在使用automap_base，因为根据业务需求我们有单独devise的模式。 我刚刚开始使用SQLAlchemy，但文档声明automap_base是declarative_base的扩展，这似乎是SQLAlchemy ORM中的典型范例，所以我相信这应该起作用。
每个Tjorriemorrie的解决scheme都不喜欢以下外键，但它只是将列与值进行匹配，并通过str（）来处理Pythontypes。 我们的值包括Python的datetime.time和decimal.Decimal类的types结果，所以它完成了工作。
希望这有助于任何路人！
你可以像这样使用SqlAlchemy的内省：
mysql = SQLAlchemy() from sqlalchemy import inspect class Contacts(mysql.Model): __tablename__ = 'CONTACTS' id = mysql.Column(mysql.Integer, primary_key=True) first_name = mysql.Column(mysql.String(128), nullable=False) last_name = mysql.Column(mysql.String(128), nullable=False) phone = mysql.Column(mysql.String(128), nullable=False) email = mysql.Column(mysql.String(128), nullable=False) street = mysql.Column(mysql.String(128), nullable=False) zip_code = mysql.Column(mysql.String(128), nullable=False) city = mysql.Column(mysql.String(128), nullable=False) def toDict(self): return { c.key: getattr(self, c.key) for c in inspect(self).mapper.column_attrs } @app.route('/contacts',methods=['GET']) def getContacts(): contacts = Contacts.query.all() contactsArr = [] for contact in contacts: contactsArr.append(contact.toDict()) return jsonify(contactsArr) @app.route('/contacts/<int:id>',methods=['GET']) def getContact(id): contact = Contacts.query.get(id) return jsonify(contact.toDict())
从这里得到灵感： 将sqlalchemy行对象转换为python字典
我知道这是一个相当古老的职位。 我采取了@SashaB给出的解决scheme，并根据我的需要进行了修改。
我添加了以下内容：
- 字段忽略列表：序列化时要忽略的字段列表
- 字段replace列表：包含字段名称的字典，在序列化时将被值replace。
- 删除方法和BaseQuery被序列化
我的代码如下：
def alchemy_json_encoder(revisit_self = False, fields_to_expand = [], fields_to_ignore = [], fields_to_replace = {}): """ Serialize SQLAlchemy result into JSon :param revisit_self: True / False :param fields_to_expand: Fields which are to be expanded for including their children and all :param fields_to_ignore: Fields to be ignored while encoding :param fields_to_replace: Field keys to be replaced by values assigned in dictionary :return: Json serialized SQLAlchemy object """ _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' and x not in fields_to_ignore]: val = obj.__getattribute__(field) # is this field method defination, or an SQLalchemy object if not hasattr(val, "__call__") and not isinstance(val, BaseQuery): field_name = fields_to_replace[field] if field in fields_to_replace else 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_name] = None continue fields[field_name] = val # a json-encodable dict return fields return json.JSONEncoder.default(self, obj) return AlchemyEncoder
希望它可以帮助别人！
我利用（太多？）字典：
def serialize(_query): #d = dictionary written to per row #D = dictionary d is written to each time, then reset #Master = dictionary of dictionaries; the id Key (int, unique from database) from D is used as the Key for the dictionary D entry in Master Master = {} D = {} x = 0 for u in _query: d = u.__dict__ D = {} for n in d.keys(): if n != '_sa_instance_state': D[n] = d[n] x = d['id'] Master[x] = D return Master
运行瓶（包括jsonify）和flask_sqlalchemy以JSON格式输出。
用jsonify（serialize（））调用函数。
适用于迄今为止我尝试过的所有SQLAlchemy查询（运行SQLite3）
