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/)
当前回答
虽然这是一篇老文章，也许我没有回答上面的问题，但我想谈谈我的连载，至少它对我有用。
我使用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/
其他回答
向任何模型添加一个_dict方法的动态方法
from sqlalchemy.inspection import inspect
def implement_as_dict(model):
if not hasattr(model,"as_dict"):
column_names=[]
imodel = inspect(model)
for c in imodel.columns:
column_names.append(c.key)
#define model.as_dict()
def as_dict(self):
d = {}
for c in column_names:
d[c] = getattr(self,c)
return d
setattr(model,"as_dict",as_dict)
#model definition
class User(Base):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
# adding as_dict definition to model
implement_as_dict(User)
然后你可以使用
user = session.query(User).filter_by(name='rick').first()
user.as_dict()
#sample output
{"id":1,"name":"rick"}
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，你可以抓取代码
如果你正在使用Flask并且只想快速查询:
def get_cats():
sql = text("select * from cat")
sql_params = {}
result = db.session.execute(sql, sql_params)
row_list = result.fetchall()
data = [dict(r) for r in row_list]
response = jsonify({
'data': [{
'categorias': data
}]
})
return response
def alc2json(row):
return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])
我想和她玩会儿代码高尔夫。
供参考:我使用automap_base，因为我们有一个根据业务需求单独设计的模式。我今天才开始使用SQLAlchemy，但是文档指出automap_base是declarative_base的扩展，这似乎是SQLAlchemy ORM中的典型范例，所以我相信这应该可以工作。
根据Tjorriemorrie的解决方案，它并没有跟随外键，而是简单地将列与值匹配，并通过str()-ing列值来处理Python类型。我们的值包括Python datetime。时间和小数。十进位类类型的结果，所以它完成了工作。
希望对路人有所帮助!
AlchemyEncoder是很棒的，但有时会失败的十进制值。这是一个改进的编码器，解决十进制问题-
class AlchemyEncoder(json.JSONEncoder):
# To serialize SQLalchemy objects
def default(self, obj):
if isinstance(obj.__class__, DeclarativeMeta):
model_fields = {}
for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:
data = obj.__getattribute__(field)
print data
try:
json.dumps(data) # this will fail on non-encodable values, like other classes
model_fields[field] = data
except TypeError:
model_fields[field] = None
return model_fields
if isinstance(obj, Decimal):
return float(obj)
return json.JSONEncoder.default(self, obj)
