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/)
在Flask下,它工作并处理datatime字段,转换类型字段
“时间”:datetime。Datetime(2018, 3, 22, 15, 40)成
“时间”:“2018-03-22 15:40:00”:
obj = {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}
# This to get the JSON body
return json.dumps(obj)
# Or this to get a response object
return jsonify(obj)
你可以像这样使用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 dict
内置序列化器因utf-8而阻塞,无法解码某些输入的无效开始字节。相反,我的答案是:
def row_to_dict(row):
temp = row.__dict__
temp.pop('_sa_instance_state', None)
return temp
def rows_to_list(rows):
ret_rows = []
for row in rows:
ret_rows.append(row_to_dict(row))
return ret_rows
@website_blueprint.route('/api/v1/some/endpoint', methods=['GET'])
def some_api():
'''
/some_endpoint
'''
rows = rows_to_list(SomeModel.query.all())
response = app.response_class(
response=jsonplus.dumps(rows),
status=200,
mimetype='application/json'
)
return response
Python 3.7+将于2023年发布
您可以将数据类装饰器添加到您的模型中,并定义一个自定义JSON序列化器,然后是JSON。转储将工作(通过向cls提供自定义编码器)。在下面的例子中,db_row是DB类的一个实例:
json.dumps(db_row, cls=models.CustomJSONEncoder)
{"id": 25, "name": "A component", "author": "Bob", "modified": "2023-02-08T11:49:15.675837"}
可以很容易地修改定制JSON序列化器,使其与任何原生JSON不可序列化的类型兼容。
models.py
from datetime import datetime
import dataclasses
import json
from sqlalchemy import Column, Integer, String, DateTime
from database import Base
@dataclasses.dataclass # <<-- add this decorator
class DB(Base):
"""Model used for SQLite database entries."""
__tablename__ = "components"
id: int = Column(Integer, primary_key=True, index=True)
name: str = Column(String)
author: str = Column(String)
modified: datetime = Column(DateTime(timezone=True), default=datetime.utcnow)
class CustomJSONEncoder(json.JSONEncoder): # <<-- Add this custom encoder
"""Custom JSON encoder for the DB class."""
def default(self, o):
if dataclasses.is_dataclass(o): # this serializes anything dataclass can handle
return dataclasses.asdict(o)
if isinstance(o, datetime): # this adds support for datetime
return o.isoformat()
return super().default(o)
为了进一步扩展它,使它适用于你在数据库中可能使用的任何不可序列化的类型,在自定义编码器类中添加另一条if语句,返回一些可序列化的东西(例如str)。
下面是一个解决方案,它允许您选择希望在输出中包含的关系。
注意:这是一个完整的重写,将dict/str作为一个参数,而不是一个列表。修复了一些东西..
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
举个关于person/family/homes/rooms的例子…把它转换成json,你只需要
json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))