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/)
class SqlToDict:
def __init__(self, data) -> None:
self.data = data
def to_timestamp(self, date):
if isinstance(date, datetime):
return int(datetime.timestamp(date))
else:
return date
def to_dict(self) -> List:
arr = []
for i in self.data:
keys = [*i.keys()]
values = [*i]
values = [self.to_timestamp(d) for d in values]
arr.append(dict(zip(keys, values)))
return arr
例如:
SqlToDict(data).to_dict()
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)。
自定义序列化和反序列化。
"from_json"(类方法)基于json数据构建一个Model对象。
“反序列化”只能在实例上调用,并将json中的所有数据合并到Model实例中。
"serialize" -递归序列化
需要__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, e.g. 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
虽然这是一篇老文章,也许我没有回答上面的问题,但我想谈谈我的连载,至少它对我有用。
我使用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/