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/)
我已经成功地使用了这个包:https://github.com/n0nSmoker/SQLAlchemy-serializer
你可以在模型上这样做:
from sqlalchemy_serializer import SerializerMixin
class SomeModel(db.Model, SerializerMixin):
...
它添加了完全递归的to_dict:
item = SomeModel.query.filter(...).one()
result = item.to_dict()
它可以让你制定规则来避免无限递归:
result = item.to_dict(rules=('-somefield', '-some_relation.nested_one.another_nested_one'))
我已经成功地使用了这个包:https://github.com/n0nSmoker/SQLAlchemy-serializer
你可以在模型上这样做:
from sqlalchemy_serializer import SerializerMixin
class SomeModel(db.Model, SerializerMixin):
...
它添加了完全递归的to_dict:
item = SomeModel.query.filter(...).one()
result = item.to_dict()
它可以让你制定规则来避免无限递归:
result = item.to_dict(rules=('-somefield', '-some_relation.nested_one.another_nested_one'))
这是一个JSONEncoder版本,它保留了模型列的顺序,只保留递归定义的列和关系字段。它还格式化了大多数不可序列化的JSON类型:
import json
from datetime import datetime
from decimal import Decimal
import arrow
from sqlalchemy.ext.declarative import DeclarativeMeta
class SQLAlchemyJSONEncoder(json.JSONEncoder):
"""
SQLAlchemy ORM JSON Encoder
If you have a "backref" relationship defined in your SQLAlchemy model,
this encoder raises a ValueError to stop an infinite loop.
"""
def default(self, obj):
if isinstance(obj, datetime):
return arrow.get(obj).isoformat()
elif isinstance(obj, Decimal):
return float(obj)
elif isinstance(obj, set):
return sorted(obj)
elif isinstance(obj.__class__, DeclarativeMeta):
for attribute, relationship in obj.__mapper__.relationships.items():
if isinstance(relationship.__getattribute__("backref"), tuple):
raise ValueError(
f'{obj.__class__} object has a "backref" relationship '
"that would cause an infinite loop!"
)
dictionary = {}
column_names = [column.name for column in obj.__table__.columns]
for key in column_names:
value = obj.__getattribute__(key)
if isinstance(value, datetime):
value = arrow.get(value).isoformat()
elif isinstance(value, Decimal):
value = float(value)
elif isinstance(value, set):
value = sorted(value)
dictionary[key] = value
for key in [
attribute
for attribute in dir(obj)
if not attribute.startswith("_")
and attribute != "metadata"
and attribute not in column_names
]:
value = obj.__getattribute__(key)
dictionary[key] = value
return dictionary
return super().default(obj)
虽然这是一篇老文章,也许我没有回答上面的问题,但我想谈谈我的连载,至少它对我有用。
我使用FastAPI,SqlAlchemy和MySQL,但我不使用orm模型;
# from sqlalchemy import create_engine
# from sqlalchemy.orm import sessionmaker
# engine = create_engine(config.SQLALCHEMY_DATABASE_URL, pool_pre_ping=True)
# SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
序列化代码
import decimal
import datetime
def alchemy_encoder(obj):
"""JSON encoder function for SQLAlchemy special classes."""
if isinstance(obj, datetime.date):
return obj.strftime("%Y-%m-%d %H:%M:%S")
elif isinstance(obj, decimal.Decimal):
return float(obj)
import json
from sqlalchemy import text
# db is SessionLocal() object
app_sql = 'SELECT * FROM app_info ORDER BY app_id LIMIT :page,:page_size'
# The next two are the parameters passed in
page = 1
page_size = 10
# execute sql and return a <class 'sqlalchemy.engine.result.ResultProxy'> object
app_list = db.execute(text(app_sql), {'page': page, 'page_size': page_size})
# serialize
res = json.loads(json.dumps([dict(r) for r in app_list], default=alchemy_encoder))
如果不行,请忽略我的回答。我在这里提到它
https://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/
我对使用(太多?)字典的看法:
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)