Wie kann man das SqlAlchemy-Ergebnis in JSON serialisieren?

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:

• Python 2.7, 3.4, 3.5 und 3.6.
• SQLAlchemy-Versionen 0.9 und höher
• serialisierung/Deserialisierung von/nach JSON, CSV, YAML und Python dict
• serialisierung/Deserialisierung von Spalten/Attributen, Beziehungen, Hybrid-Eigenschaften und Assoziations-Proxies
• aktivieren und Deaktivieren der Serialisierung für bestimmte Formate und Spalten/Beziehungen/Attribute (z. B. möchten Sie einen inboundpassword-Wert unterstützen, aber niemals einen outbound - Wert enthalten).
• vor-Serialisierung und Nachererialisierungswertverarbeitung (zur Validierung oder Typzwang)
• eine recht unkomplizierte Syntax, die sowohl Pythonic als auch nahtlos mit SQLAlchemys eigenem Ansatz übereinstimmt

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

Python 3.7+ und Flask 1.1+ können das integrierte Datenklassen -Paket verwenden

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

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

1. Feld-Ignorierliste: Eine Liste von Feldern, die bei der Serialisierung ignoriert werden sollen
2. Feldersetzungsliste: Ein Wörterbuch, das Feldnamen enthält, die beim Serialisieren durch Werte ersetzt werden.
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!