Django可以很好地自动序列化从DB返回到JSON格式的ORM模型。
如何序列化SQLAlchemy查询结果为JSON格式?
我试过jsonpickle。编码，但它编码查询对象本身。 我尝试了json.dumps(items)，但它返回
TypeError: <Product('3', 'some name', 'some desc')> is not JSON serializable
将SQLAlchemy ORM对象序列化为JSON /XML真的那么难吗?它没有任何默认序列化器吗?现在序列化ORM查询结果是非常常见的任务。
我所需要的只是返回SQLAlchemy查询结果的JSON或XML数据表示。
需要在javascript datagird中使用JSON/XML格式的SQLAlchemy对象查询结果(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)
它将忽略不可编码的字段(将它们设置为“None”)。
它不会自动展开关系(因为这可能导致自引用，并永远循环)。
递归的非循环实现
然而，如果你宁愿永远循环，你可以使用:
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)
这将编码所有的子代、子代、子代……基本上可以编码你的整个数据库。当它到达之前编码过的东西时，它会将其编码为“None”。
递归的、可能是循环的、有选择的实现
另一种选择，可能更好，是能够指定你想要展开的字段:
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)
例如，仅展开名为“parents”的SQLAlchemy字段。
其他回答
这并不是那么简单。我写了一些代码来做这件事。我还在开发中，它使用了MochiKit框架。它基本上使用代理和注册的JSON转换器在Python和Javascript之间转换复合对象。
数据库对象的浏览器端是db.js 它需要proxy.js中的基本Python代理源代码。
在Python方面，有基本代理模块。 最后是webserver.py中的SqlAlchemy对象编码器。 它还依赖于models.py文件中的元数据提取器。
你可以像这样将RowProxy转换为dict:
d = dict(row.items())
然后将其序列化为JSON(必须为datetime值等指定编码器) 如果您只想要一条记录(而不是相关记录的完整层次结构)，这并不难。
json.dumps([(dict(row.items())) for row in rs])
在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)。
扁平化实现
你可以使用这样的代码:
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)
它将忽略不可编码的字段(将它们设置为“None”)。
它不会自动展开关系(因为这可能导致自引用，并永远循环)。
递归的非循环实现
然而，如果你宁愿永远循环，你可以使用:
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)
这将编码所有的子代、子代、子代……基本上可以编码你的整个数据库。当它到达之前编码过的东西时，它会将其编码为“None”。
递归的、可能是循环的、有选择的实现
另一种选择，可能更好，是能够指定你想要展开的字段:
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)
例如，仅展开名为“parents”的SQLAlchemy字段。
我对使用(太多?)字典的看法:
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
使用flask(包括jsonify)和flask_sqlalchemy将输出打印为JSON。
使用jsonify(serialize())调用该函数。
与我迄今为止尝试过的所有SQLAlchemy查询一起工作(运行SQLite3)
