Wie serialisiere ich das SqlAlchemy-Ergebnis in JSON?

189

Django verfügt über eine gute automatische Serialisierung von ORM-Modellen, die von DB in das JSON-Format zurückgegeben wurden.

Wie serialisiere ich das SQLAlchemy-Abfrageergebnis in das JSON-Format?

Ich habe es versucht, jsonpickle.encodeaber es codiert das Abfrageobjekt selbst. Ich habe es versucht, json.dumps(items)aber es kehrt zurück

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-Serializer dafür? Heutzutage ist es eine sehr häufige Aufgabe, ORM-Abfrageergebnisse zu serialisieren.

Ich muss lediglich die JSON- oder XML-Datendarstellung des SQLAlchemy-Abfrageergebnisses zurückgeben.

Das Abfrageergebnis von SQLAlchemy-Objekten im JSON / XML-Format wird benötigt, um in Javascript-Datagird (JQGrid http://www.trirand.com/blog/ ) verwendet zu werden.

Zelid
quelle
Dies ist eine Problemumgehung, die für mich funktioniert.
Geben Sie

Antworten:

127

Eine flache Implementierung

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 konvertieren Sie dann in JSON mit:

c = YourAlchemyClass()
print json.dumps(c, cls=AlchemyEncoder)

Felder, die nicht codierbar sind, werden ignoriert (setzen Sie sie auf 'Keine').

Beziehungen werden nicht automatisch erweitert (da dies zu Selbstreferenzen führen und für immer eine Schleife bilden kann).

Eine rekursive, nicht zirkuläre Implementierung

Wenn Sie jedoch lieber für immer eine Schleife erstellen möchten, 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 codieren 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 codieren ... Im Grunde genommen möglicherweise Ihre gesamte Datenbank codieren. Wenn es etwas erreicht, das zuvor codiert wurde, wird es als "Keine" codiert.

Eine rekursive, möglicherweise kreisförmige, selektive Implementierung

Eine andere Alternative, wahrscheinlich besser, besteht darin, 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)

Zum Beispiel nur um SQLAlchemy-Felder zu erweitern, die als "Eltern" bezeichnet werden.

Sasha B.
quelle
Das ist eine großartige Antwort, aber ich bekomme eine "konnte nicht codieren" BaseQuery ", wenn es eine Beziehung mit den nicht flachen Methoden trifft, irgendwelche Ideen?
Ben Kilah
1
@SashaB Wie wäre es mit einer genaueren Ausrichtung auf Fälle, in denen sich eine Beziehung wiederholt? Zum Beispiel, wenn ich online_orderund habe address, beide mit einer Beziehung zu user, aber online_orderauch eine Beziehung zu address. Wenn ich all dies serialisieren wollte, müsste ich es addressin das aufnehmen fields_to_expand, aber ich möchte es addressaufgrund seiner Beziehung zu userund nicht redundant serialisieren online_order.
Chrispy
2
@ BenKilah Lass mich raten, du verwendest Flask-SqlAlchemy und deine Modelle erben von db.Model, nicht von Base. Wenn dies der Fall ist, ändern Sie es for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:so, dass es angezeigt wird for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata' and not x.startswith('query')]:. Denken Sie daran, dass diese Lösung verhindert, dass Sie eine Eigenschaft / Beziehung mit dem Namen 'query' haben
Pakman
genauso wie ich, aber viel komplexer. stackoverflow.com/questions/7102754/…
tyan
2
Sie können meine Lösung verwenden github.com/n0nSmoker/SQLAlchemy-serializer
n0nSmoker
269

Sie können Ihr Objekt einfach als Wörterbuch ausgeben:

class User:
   def as_dict(self):
       return {c.name: getattr(self, c.name) for c in self.__table__.columns}

Und dann benutzt du User.as_dict() um Ihr Objekt zu serialisieren.

Wie unter Konvertieren des SQLlchemie-Zeilenobjekts in Python-Diktat erläutert

charlax
quelle
2
@charlax, Wie habe ich eine DateTime repariert? Durch diese Verwendung ich 'datetime.datetime (2013, 3, 22, 16, 50, 11) nicht JSON serializable' , wenn ich json.dumps
Asken
1
Es liegt in der Verantwortung des JSONEncoderObjekts. Sie können eine Unterklasse erstellen, um Ihren eigenen Encoder für ein Objekt zu definieren, einschließlich datetime. Beachten Sie, dass Flaskbeispielsweise die sofortige Codierung von datetime in JSON (mit der neuesten Version) unterstützt wird.
Charlax
3
Wenn Sie die "deklarative" Methode von sqlalchemy verwenden, können Sie einer benutzerdefinierten Basisklasse so etwas hinzufügen - dies ist ziemlich praktisch, da Sie dann my_orm_object.toDict () für jedes ORM-Objekt aufrufen können, das Sie haben. Ebenso können Sie eine .toJSON () -Methode definieren, die Ihre toDict-Methode und einen benutzerdefinierten Encoder für die Behandlung von Datumsangaben, Blobs usw. verwendet
FredL
7
Unterstützung auch Datetime:return {c.name: unicode(getattr(self, c.name)) for c in self.__table__.columns}
Shoham
1
Dies funktioniert nicht, wenn Ihre Klassenvariablen nicht mit Ihren Spaltennamen übereinstimmen. Irgendeine Idee, wie man stattdessen die Klassennamen bekommt?
James Burke
54

Sie können einen RowProxy in ein Diktat wie das folgende konvertieren:

 d = dict(row.items())

Serialisieren Sie das dann in JSON (Sie müssen einen Encoder für Dinge wie datetimeWerte 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])
Nick Perkins
quelle
1
Dies funktioniert für meine benutzerdefinierte SQL-Abfrage mit db.engine.connect () als con: rs = con.execute (sql)
JZ.
1
Das ist viel einfacher und funktioniert. Was ist der Unterschied zwischen dieser Antwort und der akzeptierten Antwort?
Sundeep
46

Ich empfehle Marshmallow . Sie können Serializer erstellen, um Ihre Modellinstanzen mit Unterstützung für Beziehungen und verschachtelte Objekte darzustellen.

Hier ist ein abgeschnittenes Beispiel aus ihren Dokumenten. Nehmen Sie das ORM-Modell Author:

class Author(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    first = db.Column(db.String(80))
    last = db.Column(db.String(80))

Ein Marshmallow-Schema für diese Klasse ist wie folgt aufgebaut:

class AuthorSchema(Schema):
    id = fields.Int(dump_only=True)
    first = fields.Str()
    last = fields.Str()
    formatted_name = fields.Method("format_name", dump_only=True)

    def format_name(self, author):
        return "{}, {}".format(author.last, author.first)

... und so verwendet:

author_schema = AuthorSchema()
author_schema.dump(Author.query.first())

... würde eine Ausgabe wie diese erzeugen:

{
        "first": "Tim",
        "formatted_name": "Peters, Tim",
        "id": 1,
        "last": "Peters"
}

Schauen Sie sich das vollständige Flask-SQLAlchemy-Beispiel an .

Eine Bibliothek namens marshmallow-sqlalchemy SQLAlchemy und Marshmallow integriert speziell. In dieser Bibliothek Authorsieht das Schema für das oben beschriebene Modell folgendermaßen aus:

class AuthorSchema(ModelSchema):
    class Meta:
        model = Author

Durch die Integration können die Feldtypen aus den SQLAlchemy- ColumnTypen abgeleitet werden.

Marshmallow-Sqlalchemie hier.

Yasir Hantoush
quelle
12
Ich fand auch marshmallow-sqlalchemy.readthedocs.io/en/latest, was die Schemaerstellung vereinfacht
Foo L
40

Python 3.7+ und Flask 1.1+ können den Einbau-verwenden dataclasses Paket

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 __name__ == "__main__":
  users = User(email="[email protected]"), User(email="[email protected]")
  db.create_all()
  db.session.add_all(users)
  db.session.commit()
  app.run()

Die /users/Route gibt nun eine Liste der Benutzer zurück.

[
  {"email": "[email protected]", "id": 1},
  {"email": "[email protected]", "id": 2}
]

Verwandte Modelle automatisch serialisieren

@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
      }
   ]
}

Überschreiben Sie den Standard-JSON-Encoder

from flask.json import JSONEncoder


class CustomJSONEncoder(JSONEncoder):
  "Add support for serializing timedeltas"

  def default(o):
    if type(o) == datetime.timedelta:
      return str(o)
    elif type(o) == datetime.datetime:
      return o.isoformat()
    else:
      return super().default(o)

app.json_encoder = CustomJSONEncoder      
Tom
quelle
1
Das sieht nach der richtigen Art von einfach aus. Funktioniert es auch zur Deserialisierung?
Ender2050
Sie können ein Wörterbuch von analysiertem JSON in ein Modell konvertieren, indem Sie das Schlüsselwortargument entpacken:data = request.json['user']; user = User(**data)
Tom
3
Beachten Sie, dass dies id: int = Columnfunktioniert, aber id = Columnnicht. Es scheint, als müssten Sie die statische Eingabe für den JSON deklarieren, um das Feld zu serialisieren. Andernfalls erhalten Sie ein leeres {}Objekt.
Ambroise Rabier
1
Das hat bei mir funktioniert, warum ist das nicht die akzeptierte Antwort? Ich habe stundenlang mit app_context herumgespielt, damit dies mit Flask-Marshmallow funktioniert.
Nick Dat Le
1
Hat auch für mich gearbeitet. Beachten Sie, dass Sie unter Python 3.6 nur das Paket installieren möchten : pipenv install dataclasses. Und dann wird es gut funktionieren.
AleksandrH
14

Das Flask-JsonTools- Paket enthält eine Implementierung von JsonSerializableBase für Ihre Modelle.

Verwendung:

from sqlalchemy.ext.declarative import declarative_base
from flask.ext.jsontools import JsonSerializableBase

Base = declarative_base(cls=(JsonSerializableBase,))

class User(Base):
    #...

Jetzt die User Modell magisch serialisierbar.

Wenn Ihr Framework nicht Flask ist, können Sie einfach den Code abrufen

kolypto
quelle
2
Dies löst nur die Hälfte des Problems, da nur eine einzelne Zeile serialisiert wird. Wie serialisiere ich das gesamte Abfrageergebnis?
Steve Bennett
@SteveBennett verwendet das jsonapi von jsontools , um die Antwort zu codieren. Das wird automatisch das Rückgabeobjekt codieren
Tjorriemorrie
Ich habe ein sehr einfaches SQLalchemie-Modell und erhalte Folgendes: TypeError: <ORM.State-Objekt bei 0x03577A50> ist nicht JSON-serialisierbar
Matej
1
Es funktionierte schließlich, indem __json __ () für mein Modellobjekt explizit aufgerufen wurde: return my_object .__ json __ ()
Matej
Die Bibliothek funktioniert nicht mit Flask 1.0 und höher, da import flask.ext.whateverdies in Flask 1.0 nicht mehr unterstützt wird.
Adarsh ​​Madrecha
14

Aus Sicherheitsgründen sollten Sie niemals alle Felder des Modells zurückgeben. Ich ziehe es vor, sie selektiv auszuwählen.

Die json-Codierung von Flask unterstützt jetzt UUID, Datum / Uhrzeit und Beziehungen (und hinzugefügt queryund query_classfür die db.ModelKlasse flask_sqlalchemy ). 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

Damit kann ich optional eine __json__Eigenschaft hinzufügen , die die Liste der Felder zurückgibt, die ich codieren 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 meiner Ansicht @jsonapi 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"
}

]
Tjorriemorrie
quelle
Wunderschönen! Noch einmal, beweisen Sie, dass Sie manchmal nicht für jede dumme kleine Aufgabe ein dickes Paket benötigen - dass das Erlernen von DSL schwieriger sein kann, als es auf "harte" Weise zu tun. Ich habe mir viele JSON- und REST-Pakete angesehen, bevor ich hier gelandet bin. Es stimmt, das erfordert noch ein Paket, flask_jsontools (hinzufügen @jsonapizu @app.routein views.py etc), aber ich liebe die Einfachheit davon. Ich denke, es ist billig Flask hinzugefügt Datum / Uhrzeit, aber nicht Datum, also habe ich es selbst zu json_encoder.py hinzugefügt : value=...^ if isinstance(value, date):^ data[field] = datetime.combine(value, time.min).isoformat()^ else:^try:...
juanitogan
10

Sie können die Introspektion von SqlAlchemy wie folgt verwenden:

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())

Lassen Sie sich von einer Antwort hier inspirieren: Konvertieren Sie das Zeilenobjekt sqlalchemy in python dict

phico
quelle
5

Eine detailliertere 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}

Das str()ist für Python 3, also wenn Sie Python 2 verwenden, verwenden Sie unicode(). Es sollte helfen, Daten zu deserialisieren. Sie können es entfernen, wenn Sie sich nicht mit diesen befassen.

Sie können die Datenbank jetzt wie folgt 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()wird nun das Ergebnis deserialisieren. Nach der Deserialisierung kann es an json übergeben werden

jsonify(some_result)
Patrick Mutuku
quelle
3

Es ist nicht so einfach. Ich habe dazu einen Code geschrieben. Ich arbeite noch daran und es verwendet das MochiKit-Framework. Grundsätzlich werden zusammengesetzte Objekte zwischen Python und Javascript mithilfe eines Proxys und registrierter JSON-Konverter übersetzt.

Die Browserseite für Datenbankobjekte ist db.js. Sie benötigt die grundlegende Python- Proxyquelle in proxy.js .

Auf der Python-Seite befindet sich das Basis- Proxy-Modul . Dann endlich der SqlAlchemy-Objektcodierer in webserver.py . Dies hängt auch von den Metadatenextraktoren in der Datei models.py ab .

Keith
quelle
Auf den ersten Blick ziemlich kompliziert ... Was ich brauche - ist, SQLAlchemy-Objektabfrageergebnisse im JSON / XML-Format zu erhalten, um es in Javascript-Datagird (JQGrid trirand.com/blog ) zu verwenden
Zelid
Manchmal sind Probleme komplizierter, als Sie auf den ersten Blick erwarten ... Dies behandelt Objekte, die als Fremdschlüssel zurückgegeben werden, und versucht, die unendliche Rekursion zu vermeiden, die bei tief verschachtelten Beziehungen auftritt. Sie könnten jedoch wahrscheinlich einige benutzerdefinierte Abfragen schreiben, die nur Basistypen zurückgeben, und diese direkt mit simplejson serialisieren.
Keith
1
Richtig, vielleicht werde ich wirklich mit SQLAlchemy nach Diktaten fragen und die Vorteile von ORM nutzen, die nur Speicher- / Aktualisierungsaktionen ausführen.
Zelid
3

Während die ursprüngliche Frage eine Weile zurückreicht, deuten die Anzahl der Antworten hier (und meine eigenen Erfahrungen) darauf hin, dass es sich um eine nicht triviale Frage mit vielen verschiedenen Ansätzen unterschiedlicher Komplexität und unterschiedlichen Kompromissen handelt.

Aus diesem Grund habe ich die SQLAthanor- Bibliothek erstellt, die das deklarative ORM von SQLAlchemy um konfigurierbare Unterstützung für Serialisierung / De-Serialisierung erweitert, die Sie sich vielleicht ansehen möchten.

Die Bibliothek unterstützt:

  • Python 2.7, 3.4, 3.5 und 3.6.
  • SQLAlchemy-Versionen 0.9 und höher
  • Serialisierung / De-Serialisierung zu / von JSON, CSV, YAML und Python dict
  • Serialisierung / De-Serialisierung von Spalten / Attributen, Beziehungen, Hybrideigenschaften und Assoziations-Proxys
  • Aktivieren und Deaktivieren der Serialisierung für bestimmte Formate und Spalten / Beziehungen / Attribute (z. B. möchten Sie einen eingehenden password Wert unterstützen, aber niemals einen ausgehenden Wert einschließen )
  • Wertverarbeitung vor und nach der Deserialisierung (zur Validierung oder zum Typzwang)
  • Eine ziemlich einfache Syntax, die sowohl pythonisch als auch nahtlos mit dem eigenen Ansatz von SQLAlchemy übereinstimmt

Sie können die (ich hoffe!) Umfassenden Dokumente hier einsehen: https://sqlathanor.readthedocs.io/en/latest

Hoffe das hilft!

Chris Modzelewski
quelle
2

Benutzerdefinierte Serialisierung und Deserialisierung.

"from_json" (Klassenmethode) erstellt ein basierend auf JSON-Daten.

"deserialize" kann nur für eine Instanz aufgerufen werden und alle Daten von json in die Model-Instanz zusammenführen.

"serialize" - rekursive Serialisierung

Die Eigenschaft __write_only__ wird benötigt, um Nur-Schreib-Eigenschaften zu definieren (z. B. "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
Артем Федоров
quelle
2

Hier ist eine Lösung, mit der Sie die Beziehungen, die Sie in Ihre Ausgabe aufnehmen möchten, so tief auswählen können, wie Sie möchten. HINWEIS: Dies ist ein vollständiges Umschreiben, bei dem ein Diktat / Str als Argument und nicht als 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

Also für ein Beispiel mit Person / Familie / Zuhause / Zimmer ... verwandeln Sie es in json alles was Sie brauchen ist

json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))
tahoe
quelle
Ich denke, das ist in Ordnung, wenn Sie es einfach in Ihre Basisklasse einfügen, damit alle Objekte es haben. Ich überlasse Ihnen die
JSON-
Beachten Sie, dass diese Version alle Listenrelationen erhält.
Seien Sie
1
def alc2json(row):
    return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])

Ich dachte, ich würde mit diesem ein bisschen Code Golf spielen.

Zu Ihrer Information : Ich verwende automap_base da wir ein separat gestaltetes Schema haben, das den Geschäftsanforderungen entspricht. Ich habe gerade erst angefangen, SQLAlchemy zu verwenden, aber die Dokumentation besagt, dass automap_base eine Erweiterung von declarative_base ist, was das typische Paradigma im SQLAlchemy ORM zu sein scheint. Ich glaube, dass dies funktionieren sollte.

Das Befolgen von Fremdschlüsseln gemäß der Lösung von Tjorriemorrie macht keine Lust , aber es ordnet Spalten einfach Werten zu und behandelt Python-Typen, indem es die Spaltenwerte str () -. Unsere Werte bestehen aus Python datetime.time und decimal.Decimal-Klassentypergebnissen, damit die Aufgabe erledigt wird.

Hoffe das hilft allen Passanten!

hpatel71
quelle
1

Ich weiß, dass dies ein ziemlich älterer Beitrag ist. Ich nahm die von @SashaB gegebene Lösung und modifizierte sie gemäß meinen Anforderungen.

Ich habe folgende Dinge hinzugefügt:

  1. Feld-Ignorierliste: Eine Liste von Feldern, die beim Serialisieren ignoriert werden sollen
  2. Feldersetzungsliste: Ein Wörterbuch mit Feldnamen, die beim Serialisieren durch Werte ersetzt werden sollen.
  3. Methoden entfernt und BaseQuery wird serialisiert

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!

srahul07
quelle
1

Verwenden Sie den in SQLAlchemy integrierten Serializer :

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, denken Sie daran, das Objekt mithilfe von von der aktuellen Sitzung zu trennen session.expunge(obj). Um es wieder anzubringen, tun Sie es einfach session.add(obj).

chribsen
quelle
Nifty, wird aber nicht in JSON konvertiert.
Blakev
2
Informationen zur JSON-Serialisierung finden Sie unter Marshmallow-Sqlalchemy . Auf jeden Fall die beste Lösung, wenn Sie Objekte Clients aussetzen. marshmallow-sqlalchemy.readthedocs.io
chribsen
Das Serializer-Modul ist nur für Abfragestrukturen geeignet. Es wird nicht benötigt für: Instanzen von benutzerdefinierten Klassen. Diese enthalten im typischen Fall keine Verweise auf Engines, Sessions oder Ausdruckskonstrukte und können direkt serialisiert werden.
Thomasd
1

Der folgende Code serialisiert das Ergebnis von sqlalchemy 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ß nennen,

def all_products():
    all_products = Products.query.all()
    return to_array(all_products)
Chirag Vora
quelle
1

Der AlchemyEncoder ist wunderbar, schlägt aber manchmal mit Dezimalwerten fehl. 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)
Essig Agarwal
quelle
1

Wenn Sie mit sqlalchemy eine Verbindung zu einer Datenbank herstellen, ist dies eine einfache Lösung, die in hohem Maße konfigurierbar ist. Verwenden Sie Pandas.

import pandas as pd
import sqlalchemy

#sqlalchemy engine configuration
engine = sqlalchemy.create_engine....

def my_function():
  #read in from sql directly into a pandas dataframe
  #check the pandas documentation for additional config options
  sql_DF = pd.read_sql_table("table_name", con=engine)

  # "orient" is optional here but allows you to specify the json formatting you require
  sql_json = sql_DF.to_json(orient="index")

  return sql_json
chandler9
quelle
0

Unter Flask funktioniert dies und verarbeitet Datenzeitfelder, wobei ein Feld vom Typ
'time': datetime.datetime(2018, 3, 22, 15, 40)in Folgendes umgewandelt wird
"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)
belvederef
quelle
0

Die eingebauten Serializer-Drosseln mit utf-8 können für einige Eingaben kein ungültiges Startbyte dekodieren. 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
RobotHumans
quelle
0

Vielleicht können Sie eine Klasse wie diese verwenden

from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy import Table


class Custom:
    """Some custom logic here!"""

    __table__: Table  # def for mypy

    @declared_attr
    def __tablename__(cls):  # pylint: disable=no-self-argument
        return cls.__name__  # pylint: disable= no-member

    def to_dict(self) -> Dict[str, Any]:
        """Serializes only column data."""
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

Base = declarative_base(cls=Custom)

class MyOwnTable(Base):
    #COLUMNS!

Damit haben alle Objekte die to_dictMethode

Andrex
quelle
0

Während ich einige unformatierte SQL- und undefinierte Objekte verwendete, cursor.descriptionschien die Verwendung das zu bekommen, wonach ich suchte:

with connection.cursor() as cur:
    print(query)
    cur.execute(query)
    for item in cur.fetchall():
        row = {column.name: item[i] for i, column in enumerate(cur.description)}
        print(row)
jmunsch
quelle
0
step1:
class CNAME:
   ...
   def as_dict(self):
       return {item.name: getattr(self, item.name) for item in self.__table__.columns}

step2:
list = []
for data in session.query(CNAME).all():
    list.append(data.as_dict())

step3:
return jsonify(list)
邢 烽 朔
quelle
2
Code-Dumps ohne Erklärung sind selten hilfreich. Beim Stapelüberlauf geht es um das Lernen und nicht um das blinde Kopieren und Einfügen von Snippets. Bitte bearbeiten Sie Ihre Frage und erklären Sie, wie es besser funktioniert als das, was das OP bereitgestellt hat.
Chris
-2

Ich nehme (zu viele?) Wörterbücher:

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

Wird mit flask (einschließlich jsonify) und flask_sqlalchemy ausgeführt, um Ausgaben als JSON zu drucken.

Rufen Sie die Funktion mit jsonify (serialize ()) auf.

Funktioniert mit allen SQLAlchemy-Abfragen, die ich bisher ausprobiert habe (mit SQLite3)

Sawyer Brooks
quelle