Django hat einige gute automatische Serialisierung von ORM-Modellen, die vom DB in das JSON-Format zurückgegeben werden.
Wie wird das SQLAlchemy-Abfrageergebnis in das JSON-Format serialisiert?
Ich habe jsonpickle.encode
ausprobiert, aber es codiert das Abfrageobjekt selbst . Ich habe json.dumps(items)
versucht, aber es wird zurückgegeben
TypeError: <Product('3', 'some name', 'some desc')> is not JSON serializable
Ist es wirklich so schwierig, SQLAlchemy ORM-Objekte in JSON/XML zu serialisieren? Gibt es keinen Standard-Serialisierer dafür? Heutzutage ist es sehr üblich, ORM-Abfrageergebnisse zu serialisieren.
Was ich brauche, ist nur JSON- oder XML-Datendarstellung des SQLAlchemy-Abfrageergebnisses zurückzugeben.
Die Abfrageergebnisse der SQLAlchemy-Objekte im JSON/XML-Format müssen in Javascript Datagird verwendet werden (JQGrid http://www.trirand.com/blog/ ).
Sie könnten so etwas verwenden:
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)
und dann in JSON konvertieren mit:
c = YourAlchemyClass()
print json.dumps(c, cls=AlchemyEncoder)
Felder, die nicht codierbar sind, werden ignoriert (setzen Sie sie auf 'None').
Beziehungen werden nicht automatisch erweitert (da dies zu Selbstreferenzen führen und für immer in einer Schleife bleiben kann).
Wenn Sie jedoch lieber für immer eine Schleife machen, können Sie Folgendes verwenden:
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
Und dann kodieren Sie Objekte mit:
print json.dumps(e, cls=new_alchemy_encoder(), check_circular=False)
Dies würde alle Kinder und alle ihre Kinder und alle ihre Kinder verschlüsseln. Potentiell möglicherweise die gesamte Datenbank verschlüsseln. Wenn es etwas erreicht, das zuvor codiert wurde, wird es als "None" codiert.
Eine andere, wahrscheinlich bessere Alternative ist, die Felder angeben zu können, die Sie erweitern möchten:
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
Sie können es jetzt aufrufen mit:
print json.dumps(e, cls=new_alchemy_encoder(False, ['parents']), check_circular=False)
Um nur SQLAlchemy-Felder zu erweitern, die beispielsweise als "Eltern" bezeichnet werden.
Sie können Ihr Objekt einfach als Diktat ausgeben:
class User:
def as_dict(self):
return {c.name: getattr(self, c.name) for c in self.__table__.columns}
Und dann verwenden Sie User.as_dict (), um Ihr Objekt zu serialisieren.
Wie in Konvertieren des sqlalchemy-Zeilenobjekts in Python-Dikt.
Sie können ein RowProxy wie folgt in ein Diktat konvertieren:
d = dict(row.items())
Dann serialisieren Sie das zu JSON (Sie müssen einen Encoder für Dinge wie datetime
-Werte angeben) Es ist nicht so schwer, wenn Sie nur einen Datensatz (und keine vollständige Hierarchie verwandter Datensätze) möchten.
json.dumps([(dict(row.items())) for row in rs])
Ich empfehle die Verwendung einer aktuellen oberflächlichen Bibliothek Marshmallow . Sie können damit Serialisierer erstellen, die Ihre Modellinstanzen mit Unterstützung für Beziehungen und verschachtelte Objekte darstellen.
Schauen Sie sich theier SQLAlchemy Example an.
Flask-JsonTools package hat eine Implementierung von JsonSerializableBase Basisklasse für Ihre Modelle.
Verwendungszweck:
from sqlalchemy.ext.declarative import declarative_base
from flask.ext.jsontools import JsonSerializableBase
Base = declarative_base(cls=(JsonSerializableBase,))
class User(Base):
#...
Nun ist das Modell User
magisch serialisierbar.
Wenn Ihr Framework nicht Flask ist, können Sie einfach den Code nehmen
Aus Sicherheitsgründen sollten Sie niemals alle Felder des Modells zurückgeben. Ich ziehe es vor, sie selektiv auszuwählen.
Die json-Kodierung von Flask unterstützt jetzt UUID, datetime und Beziehungen (und fügte query
und query_class
für die Klasse flask_sqlalchemy db.Model
hinzu). Ich habe den Encoder wie folgt aktualisiert:
app/json_encoder.py
from sqlalchemy.ext.declarative import DeclarativeMeta
from flask import json
class AlchemyEncoder(json.JSONEncoder):
def default(self, o):
if isinstance(o.__class__, DeclarativeMeta):
data = {}
fields = o.__json__() if hasattr(o, '__json__') else dir(o)
for field in [f for f in fields if not f.startswith('_') and f not in ['metadata', 'query', 'query_class']]:
value = o.__getattribute__(field)
try:
json.dumps(value)
data[field] = value
except TypeError:
data[field] = None
return data
return json.JSONEncoder.default(self, o)
app/__init__.py
# json encoding
from app.json_encoder import AlchemyEncoder
app.json_encoder = AlchemyEncoder
Hiermit kann ich optional eine __json__
-Eigenschaft hinzufügen, die die Liste der Felder zurückgibt, die ich kodieren möchte:
app/models.py
class Queue(db.Model):
id = db.Column(db.Integer, primary_key=True)
song_id = db.Column(db.Integer, db.ForeignKey('song.id'), unique=True, nullable=False)
song = db.relationship('Song', lazy='joined')
type = db.Column(db.String(20), server_default=u'audio/mpeg')
src = db.Column(db.String(255), nullable=False)
created_at = db.Column(db.DateTime, server_default=db.func.now())
updated_at = db.Column(db.DateTime, server_default=db.func.now(), onupdate=db.func.now())
def __init__(self, song):
self.song = song
self.src = song.full_path
def __json__(self):
return ['song', 'src', 'type', 'created_at']
Ich füge @jsonapi meiner Ansicht hinzu, gebe die Ergebnisliste zurück und meine Ausgabe lautet dann wie folgt:
[
{
"created_at": "Thu, 23 Jul 2015 11:36:53 GMT",
"song":
{
"full_path": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3",
"id": 2,
"path_name": "Audioslave/Audioslave [2002]/1 Cochise.mp3"
},
"src": "/static/music/Audioslave/Audioslave [2002]/1 Cochise.mp3",
"type": "audio/mpeg"
}
]
Sie können die Introspektion von SqlAlchemy wie folgt verwenden:
mysql = SQLAlchemy()
from sqlalchemy import inspect
class Contacts(mysql.Model):
__table= '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())
Lassen Sie sich von einer Antwort hier inspirieren: Konvertiere ein sqlalchemy-Zeilenobjekt in ein Python-Dikt
Eine ausführlichere Erklärung. Fügen Sie in Ihrem Modell Folgendes hinzu:
def as_dict(self):
return {c.name: str(getattr(self, c.name)) for c in self.__table__.columns}
str()
ist für Python 3, wenn Sie Python 2 verwenden, verwenden Sie unicode()
. Es sollte helfen, Termine zu deserialisieren. Sie können es entfernen, wenn Sie sich nicht damit befassen.
Sie können jetzt die Datenbank auf diese Weise abfragen
some_result = User.query.filter_by(id=current_user.id).first().as_dict()
First()
wird benötigt, um seltsame Fehler zu vermeiden. as_dict()
deserialisiert jetzt das Ergebnis. Nach der Deserialisierung kann Json verwendet werden
jsonify(some_result)
Es ist nicht so einfach. Ich habe dazu einen Code geschrieben. Ich arbeite noch daran und es verwendet das MochiKit-Framework. Es übersetzt im Wesentlichen zusammengesetzte Objekte zwischen Python und Javascript mithilfe eines Proxy und registrierter JSON-Konverter.
Die Browserseite für Datenbankobjekte ist db.js Sie benötigt die grundlegende Python-Proxy-Quelle in proxy.js .
Auf der Python-Seite gibt es das Basismodul proxy . Dann schließlich den SqlAlchemy-Objektcodierer in webserver.py . Er hängt auch von Metadaten-Extraktoren in models.py ab Datei.
Hier ist eine Lösung, mit der Sie die Relationen auswählen können, die Sie in Ihre Ausgabe aufnehmen möchten, so tief wie Sie möchten .. HINWEIS: Dies ist ein komplettes Neuschreiben, bei dem ein dict/str als Argument anstelle einer Liste verwendet wird . behebt einige Sachen ..
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
so zum beispiel mit person/family/homes/rooms ... alles, was sie brauchen, ist json
json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))
Während die ursprüngliche Frage schon eine Weile zurückliegt, deutet die Anzahl der Antworten (und meine eigenen Erfahrungen) darauf hin, dass es sich um eine nicht triviale Frage handelt, bei der es viele unterschiedliche Ansätze unterschiedlicher Komplexität mit unterschiedlichen Kompromissen gibt.
Aus diesem Grund habe ich die Bibliothek " SQLAthanor " entwickelt, mit der der deklarative ORM von SQLAlchemy um konfigurierbare Serialisierungs-/Deserialisierungs-Unterstützung erweitert wird, die Sie vielleicht näher betrachten möchten.
Die Bibliothek unterstützt:
dict
password
-Wert unterstützen, aber niemals einen outbound - Wert enthalten).Sie können die (ich hoffe!) Umfassenden Dokumente hier einsehen: https://sqlathanor.readthedocs.io/de/latest
Hoffe das hilft!
Kundenspezifische Serialisierung und Deserialisierung.
"from_json" (Klassenmethode) erstellt ein Model-Objekt basierend auf Json-Daten.
"deserialize" konnte nur bei einer Instanz aufgerufen werden und alle Daten von Json in die Model-Instanz zusammenführen.
"serialize" - rekursive Serialisierung
_WRITE_ONLY_Eigenschaft wird benötigt, um nur schreibende Eigenschaften zu definieren ("password_hash" zum Beispiel).
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
def alc2json(row):
return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])
Ich dachte, ich würde mit diesem einen kleinen Code-Golf spielen.
Zu Ihrer Information: Ich verwende automap_base , da wir ein gemäß den Geschäftsanforderungen getrenntes Schema haben. Ich habe gerade angefangen, SQLAlchemy heute zu verwenden, aber die Dokumentation besagt, dass automap_base eine Erweiterung von declarative_base ist, die das typische Paradigma im SQLAlchemy-ORM zu sein scheint.
Mit folgenden Fremdschlüsseln pro Tjorriemorrie 's Lösung wird es nicht sonderlich interessant, es werden jedoch nur Spalten mit Werten abgeglichen und Python-Typen durch str () - Werte der Spalten behandelt. Unsere Werte bestehen aus den Python-Werten datetime.time und decimal.Decimal, damit der Job erledigt wird.
Hoffe, das hilft Passanten!
der folgende Code serialisiert das sqlalchemy-Ergebnis in json.
import json
from collections import OrderedDict
def asdict(self):
result = OrderedDict()
for key in self.__mapper__.c.keys():
if getattr(self, key) is not None:
result[key] = str(getattr(self, key))
else:
result[key] = getattr(self, key)
return result
def to_array(all_vendors):
v = [ ven.asdict() for ven in all_vendors ]
return json.dumps(v)
Spaß anrufen,
def all_products():
all_products = Products.query.all()
return to_array(all_products)
Der AlchemyEncoder ist wunderbar, versagt aber manchmal mit Dezimalwerten. Hier ist ein verbesserter Encoder, der das Dezimalproblem löst -
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)
Verwenden Sie den eingebauten Serializer in SQLAlchemy:
from sqlalchemy.ext.serializer import loads, dumps
obj = MyAlchemyObject()
# serialize object
serialized_obj = dumps(obj)
# deserialize object
obj = loads(serialized_obj)
Wenn Sie das Objekt zwischen Sitzungen übertragen, vergessen Sie nicht, das Objekt mit session.expunge(obj)
..__ von der aktuellen Sitzung zu trennen. Um es erneut anzuhängen, führen Sie einfach session.add(obj)
aus.
Unter Flask funktioniert dies und behandelt Datumsfelder, wobei ein Feld vom Typ umgewandelt wird'time': datetime.datetime(2018, 3, 22, 15, 40)
in"time": "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)
Die eingebauten Serializer-Drosseln mit utf-8 können für einige Eingänge kein ungültiges Startbyte decodieren. Stattdessen ging ich mit:
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
from dataclasses import dataclass
from datetime import datetime
from flask import Flask, jsonify
from flask_sqlalchemy import SQLAlchemy
app = Flask(__name__)
db = SQLAlchemy(app)
@dataclass
class User(db.Model):
id: int
email: str
id = db.Column(db.Integer, primary_key=True, auto_increment=True)
email = db.Column(db.String(200), unique=True)
@app.route('/users/')
def users():
users = User.query.all()
return jsonify(users)
if __== "__main__":
users = User(email="[email protected]"), User(email="[email protected]")
db.create_all()
db.session.add_all(users)
db.session.commit()
app.run()
Die Route /users/
gibt nun eine Liste von Benutzern zurück.
[
{"email": "[email protected]", "id": 1},
{"email": "[email protected]", "id": 2}
]
@dataclass
class Account(db.Model):
id: int
users: User
id = db.Column(db.Integer)
users = db.relationship(User) # User model would need a db.ForeignKey field
Die Antwort von jsonify(account)
wäre dies.
{
"id":1,
"users":[
{
"email":"[email protected]",
"id":1
},
{
"email":"[email protected]",
"id":2
}
]
}
from flask.json import JSONEncoder
class CustomJSONEncoder(JSONEncoder):
"Add support for serializing timedeltas"
def default(o):
if type(o) == datetime.timedelta:
return str(o)
else:
return super().default(o)
app.json_encoder = CustomJSONEncoder
Ich weiß, dass dies ein älterer Beitrag ist. Ich nahm die von @SashaB gegebene Lösung und modifizierte sie entsprechend meinem Bedarf.
Ich habe folgende Dinge hinzugefügt:
Mein Code lautet wie folgt:
def alchemy_json_encoder(revisit_self = False, fields_to_expand = [], fields_to_ignore = [], fields_to_replace = {}):
"""
Serialize SQLAlchemy result into JSon
:param revisit_self: True / False
:param fields_to_expand: Fields which are to be expanded for including their children and all
:param fields_to_ignore: Fields to be ignored while encoding
:param fields_to_replace: Field keys to be replaced by values assigned in dictionary
:return: Json serialized SQLAlchemy object
"""
_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' and x not in fields_to_ignore]:
val = obj.__getattribute__(field)
# is this field method defination, or an SQLalchemy object
if not hasattr(val, "__call__") and not isinstance(val, BaseQuery):
field_name = fields_to_replace[field] if field in fields_to_replace else 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_name] = None
continue
fields[field_name] = val
# a json-encodable dict
return fields
return json.JSONEncoder.default(self, obj)
return AlchemyEncoder
Hoffe es hilft jemandem!