jsonifizieren Sie eine SQLAlchemy-Ergebnismenge in Flask

Ich hatte das gleiche Bedürfnis, in json zu serialisieren. Schauen Sie sich diese Frage an. Es wird gezeigt, wie Spalten programmgesteuert erkannt werden. Daraus habe ich den folgenden Code erstellt. Es funktioniert für mich und ich werde es in meiner Web-App verwenden. Viel Spaß beim Codieren!

def to_json(inst, cls):
    """
    Jsonify the sql alchemy query result.
    """
    convert = dict()
    # add your coversions for things like datetime's 
    # and what-not that aren't serializable.
    d = dict()
    for c in cls.__table__.columns:
        v = getattr(inst, c.name)
        if c.type in convert.keys() and v is not None:
            try:
                d[c.name] = convert[c.type](v)
            except:
                d[c.name] = "Error:  Failed to covert using ", str(convert[c.type])
        Elif v is None:
            d[c.name] = str()
        else:
            d[c.name] = v
    return json.dumps(d)

class Person(base):
    __table= 'person'
    id = Column(Integer, Sequence('person_id_seq'), primary_key=True)
    first_name = Column(Text)
    last_name = Column(Text)
    email = Column(Text)

    @property
    def json(self):
        return to_json(self, self.__class__)

Folgendes ist für mich normalerweise ausreichend:

Ich erstelle einen Serialisierungsmix, den ich mit meinen Modellen verwende. Die Serialisierungsfunktion ruft im Wesentlichen alle Attribute ab, die der SQLAlchemy-Inspektor verfügbar macht, und schreibt sie in ein Diktat.

from sqlalchemy.inspection import inspect

class Serializer(object):

    def serialize(self):
        return {c: getattr(self, c) for c in inspect(self).attrs.keys()}

    @staticmethod
    def serialize_list(l):
        return [m.serialize() for m in l]

Jetzt muss nur noch das SQLAlchemy-Modell mit der mixin-Klasse Serializer erweitert werden.

Wenn es Felder gibt, die Sie nicht verfügbar machen möchten oder die eine spezielle Formatierung erfordern, überschreiben Sie einfach die Funktion serialize() in der Modellunterklasse.

class User(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    username = db.Column(db.String)
    password = db.Column(db.String)

    # ...

    def serialize(self):
        d = Serializer.serialize(self)
        del d['password']
        return d

In Ihren Steuerungen müssen Sie lediglich die Funktion serialize() (oder serialize_list(l), wenn die Abfrage eine Liste ergibt) für die Ergebnisse aufrufen:

def get_user(id):
    user = User.query.get(id)
    return json.dumps(user.serialize())

def get_users():
    users = User.query.all()
    return json.dumps(User.serialize_list(users))

Hier ist mein Ansatz: https://github.com/n0nSmoker/SQLAlchemy-serializer

pip installieren SQLAlchemy-Serializer

Sie können Ihrem Modell ganz einfach Mixin hinzufügen und nicht nur die .to_dict () -Methode für seine Instanz aufrufen

Sie können auch Ihr eigenes Mixin auf Basis von SerializerMixin schreiben

Ok, ich habe ein paar Stunden daran gearbeitet und etwas entwickelt, von dem ich glaube, dass es die bisher pythonischste Lösung ist. Die folgenden Code-Schnipsel sind Python3, sollten aber nicht zu schmerzhaft sein, wenn Sie Backport benötigen.

Das erste, was wir tun werden, ist, mit einem Mixin zu beginnen, das Ihre DB-Models dazu bringt, sich wie dicts zu verhalten:

from sqlalchemy.inspection import inspect

class ModelMixin:
    """Provide dict-like interface to db.Model subclasses."""

    def __getitem__(self, key):
        """Expose object attributes like dict values."""
        return getattr(self, key)

    def keys(self):
        """Identify what db columns we have."""
        return inspect(self).attrs.keys()

Jetzt definieren wir unser Modell und erben das Mixin:

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)
    # etc ...

Das ist alles, was es braucht, um eine Instanz von MyModel() an dict() zu übergeben und eine echte Instanz von dict zu erhalten, was uns ziemlich weit bringt zum Verständnis von jsonify(). Als nächstes müssen wir JSONEncoder erweitern, um den Rest des Weges zu schaffen:

from flask.json import JSONEncoder
from contextlib import suppress

class MyJSONEncoder(JSONEncoder):
    def default(self, obj):
        # Optional: convert datetime objects to ISO format
        with suppress(AttributeError):
            return obj.isoformat()
        return dict(obj)

app.json_encoder = MyJSONEncoder

Bonuspunkte: Wenn Ihr Modell berechnete Felder enthält (dh, Sie möchten, dass Ihre JSON-Ausgabe Felder enthält, die nicht in der Datenbank gespeichert sind), ist dies ebenfalls ganz einfach. Definieren Sie einfach Ihre berechneten Felder als @property Und erweitern Sie die keys() -Methode wie folgt:

class MyModel(db.Model, ModelMixin):
    id = db.Column(db.Integer, primary_key=True)
    foo = db.Column(...)
    bar = db.Column(...)

    @property
    def computed_field(self):
        return 'this value did not come from the db'

    def keys(self):
        return super().keys() + ['computed_field']

Jetzt ist es trivial zu jsonifizieren:

@app.route('/whatever', methods=['GET'])
def whatever():
    return jsonify(dict(results=MyModel.query.all()))

Für eine flache Abfrage (keine Verknüpfungen) können Sie dies tun

@app.route('/results/')
def results():
    data = Table.query.all()
    result = [d.__dict__ for d in data]
    return jsonify(result=result)

und wenn Sie nur bestimmte Spalten aus der Datenbank zurückgeben möchten, können Sie dies tun

@app.route('/results/')
def results():
    cols = ['id', 'url', 'shipping']
    data = Table.query.all()
    result = [{col: getattr(d, col) for col in cols} for d in data]
    return jsonify(result=result)

Wenn Sie flask-restful Verwenden, können Sie Marschall verwenden:

from flask.ext.restful import Resource, fields, marshal

topic_fields = {
    'title':   fields.String,
    'content': fields.String,
    'uri':     fields.Url('topic'),
    'creator': fields.String,
    'created': fields.DateTime(dt_format='rfc822')
}

class TopicListApi(Resource):
    def get(self):
        return {'topics': [marshal(topic, topic_fields) for topic in DbTopic.query.all()]}

Sie müssen explizit auflisten, was Sie zurückgeben und welcher Typ es ist, was ich für eine API sowieso bevorzuge. Die Serialisierung ist leicht erledigt (keine Notwendigkeit für jsonify), Termine sind auch kein Problem. Beachten Sie, dass der Inhalt für das Feld uri automatisch basierend auf dem Endpunkt topic und der ID generiert wird.

Ich habe dieses Problem den größten Teil eines Tages untersucht und habe mir Folgendes ausgedacht (Dank an https://stackoverflow.com/a/5249214/196358 für den Hinweis mich in diese Richtung).

(Hinweis: Ich verwende flask-sqlalchemy, daher unterscheidet sich mein Modelldeklarationsformat ein wenig von der reinen sqlalchemy).

In meinem models.py Datei:

import json

class Serializer(object):
  __public__ = None
  "Must be implemented by implementors"

  def to_serializable_dict(self):
    dict = {}
    for public_key in self.__public__:
      value = getattr(self, public_key)
      if value:
        dict[public_key] = value
    return dict

class SWEncoder(json.JSONEncoder):
  def default(self, obj):
    if isinstance(obj, Serializer):
      return obj.to_serializable_dict()
    if isinstance(obj, (datetime)):
      return obj.isoformat()
    return json.JSONEncoder.default(self, obj)


def SWJsonify(*args, **kwargs):
  return current_app.response_class(json.dumps(dict(*args, **kwargs), cls=SWEncoder, indent=None if request.is_xhr else 2), mimetype='application/json')
  # stolen from https://github.com/mitsuhiko/flask/blob/master/flask/helpers.py

und alle meine Modellobjekte sehen so aus:

class User(db.Model, Serializer):
  __public__ = ['id','username']
  ... field definitions ...

In meinen Ansichten rufe ich SWJsonify überall dort auf, wo ich Jsonify aufgerufen hätte:

@app.route('/posts')
def posts():
  posts = Post.query.limit(PER_PAGE).all()
  return SWJsonify({'posts':posts })

Scheint ziemlich gut zu funktionieren. Auch bei Beziehungen. Ich bin damit noch nicht weit gekommen, also YMMV, aber bisher fühlt es sich für mich ziemlich "richtig" an.

Vorschläge sind willkommen.

Hier ist meine Antwort, wenn Sie die deklarative Basis verwenden (mit Hilfe einiger bereits geposteter Antworten):

# in your models definition where you define and extend declarative_base()
from sqlalchemy.ext.declarative import declarative_base
...
Base = declarative_base()
Base.query = db_session.query_property()
...

# define a new class (call "Model" or whatever) with an as_dict() method defined
class Model():
    def as_dict(self):
        return { c.name: getattr(self, c.name) for c in self.__table__.columns }

# and extend both the Base and Model class in your model definition, e.g.
class Rating(Base, Model):
    ____table= 'rating'
    id = db.Column(db.Integer, primary_key=True)
    fullurl = db.Column(db.String())
    url = db.Column(db.String())
    comments = db.Column(db.Text)
    ...

# then after you query and have a resultset (rs) of ratings
rs = Rating.query.all()

# you can jsonify it with
s = json.dumps([r.as_dict() for r in rs], default=alchemyencoder)
print (s)

# or if you have a single row
r = Rating.query.first()

# you can jsonify it with
s = json.dumps(r.as_dict(), default=alchemyencoder)

# you will need this alchemyencoder where your are calling json.dumps to handle datetime and decimal format
# credit to Joonas @ http://codeandlife.com/2014/12/07/sqlalchemy-results-to-json-the-easy-way/
def alchemyencoder(obj):
    """JSON encoder function for SQLAlchemy special classes."""
    if isinstance(obj, datetime.date):
        return obj.isoformat()
    Elif isinstance(obj, decimal.Decimal):
        return float(obj)

Hier ist eine Möglichkeit, jeder Klasse eine as_dict () -Methode hinzuzufügen, sowie jede andere Methode, die Sie für jede einzelne Klasse haben möchten. Ich bin mir nicht sicher, ob dies der gewünschte Weg ist oder nicht, aber es funktioniert ...

class Base(object):
    def as_dict(self):
        return dict((c.name,
                     getattr(self, c.name))
                     for c in self.__table__.columns)


Base = declarative_base(cls=Base)

Ich habe nach etwas ähnlichem wie dem in ActiveRecord to_json verwendeten Rails) - Ansatz gesucht und etwas Ähnliches mit diesem Mixin implementiert, nachdem ich mit anderen Vorschlägen nicht zufrieden war. Es behandelt verschachtelte Modelle und schließt Attribute der obersten Ebene ein oder aus oder verschachtelte Modelle.

class Serializer(object):

    def serialize(self, include={}, exclude=[], only=[]):
        serialized = {}
        for key in inspect(self).attrs.keys():
            to_be_serialized = True
            value = getattr(self, key)
            if key in exclude or (only and key not in only):
                to_be_serialized = False
            Elif isinstance(value, BaseQuery):
                to_be_serialized = False
                if key in include:
                    to_be_serialized = True
                    nested_params = include.get(key, {})
                    value = [i.serialize(**nested_params) for i in value]

            if to_be_serialized:
                serialized[key] = value

        return serialized

Um die BaseQuery serialisierbar zu machen, habe ich BaseQuery erweitert

class SerializableBaseQuery(BaseQuery):

    def serialize(self, include={}, exclude=[], only=[]):
        return [m.serialize(include, exclude, only) for m in self]

Für folgende Modelle

class ContactInfo(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey('user.id'))
    full_name = db.Column(db.String())
    source = db.Column(db.String())
    source_id = db.Column(db.String())

    email_addresses = db.relationship('EmailAddress', backref='contact_info', lazy='dynamic')
    phone_numbers = db.relationship('PhoneNumber', backref='contact_info', lazy='dynamic')


class EmailAddress(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    email_address = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey('contact_info.id'))


class PhoneNumber(db.Model, Serializer):
    id = db.Column(db.Integer, primary_key=True)
    phone_number = db.Column(db.String())
    type = db.Column(db.String())
    contact_info_id = db.Column(db.Integer, db.ForeignKey('contact_info.id'))

    phone_numbers = db.relationship('Invite', backref='phone_number', lazy='dynamic')

Sie könnten so etwas tun

@app.route("/contact/search", methods=['GET'])
def contact_search():
    contact_name = request.args.get("name")
    matching_contacts = ContactInfo.query.filter(ContactInfo.full_name.like("%{}%".format(contact_name)))

    serialized_contact_info = matching_contacts.serialize(
        include={
            "phone_numbers" : {
                "exclude" : ["contact_info", "contact_info_id"]
            },
            "email_addresses" : {
                "exclude" : ["contact_info", "contact_info_id"]
            }
        }
    )

    return jsonify(serialized_contact_info)

Flask-Restful0.3.6das Request Parsing Marshmallow empfehlen

Marshmallow ist eine ORM/ODM/Framework-unabhängige Bibliothek zum Konvertieren komplexer Datentypen, z. B. Objekte, in native Python Datentypen.

Unten sehen Sie ein einfaches Marshmallow Beispiel.

from Marshmallow import Schema, fields

class UserSchema(Schema):
    name = fields.Str()
    email = fields.Email()
    created_at = fields.DateTime()

from Marshmallow import pprint

user = User(name="Monty", email="[email protected]")
schema = UserSchema()
result = schema.dump(user)
pprint(result)
# {"name": "Monty",
#  "email": "[email protected]",
#  "created_at": "2014-08-17T14:54:16.049594+00:00"}

Die Kernfunktionen enthalten

Schemata deklarieren
Objekte serialisieren ("Dumping")
Objekte deserialisieren ("Laden")
Umgang mit Sammlungen von Objekten
Validierung
Angeben von Attributnamen
Angeben von Serialisierungs-/Deserialisierungsschlüsseln
Refactoring: Implizite Felderstellung
Ausgabe bestellen
Felder "Nur Lesen" und "Nur Schreiben"
Standardwerte für Serialisierung/Deserialisierung festlegen
Schachtelschemata
Benutzerdefinierte Felder

Ich arbeitete mit einer SQL-Abfrage defaultdict von Listen von RowProxy-Objekten mit dem Namen jobDict. Es dauerte eine Weile, bis ich herausgefunden hatte, welcher Typ die Objekte waren.

Dies war ein sehr einfacher und schneller Weg, um ein sauberes Problem mit jsonEncoding zu lösen, indem die Zeile in eine Liste geschrieben und das Diktat zunächst mit dem Wert list definiert wurde.

    jobDict = defaultdict(list)
    def set_default(obj):
        # trickyness needed here via import to know type
        if isinstance(obj, RowProxy):
            return list(obj)
        raise TypeError


    jsonEncoded = json.dumps(jobDict, default=set_default)

Ich möchte nur meine Methode hinzufügen, um dies zu tun.

definieren Sie einfach einen benutzerdefinierten JSON-Encoder, um Ihre DB-Modelle zu serialisieren.

class ParentEncoder(json.JSONEncoder):
    def default(self, obj):
        # convert object to a dict
        d = {}
        if isinstance(obj, Parent):
            return {"id": obj.id, "name": obj.name, 'children': list(obj.child)}
        if isinstance(obj, Child):
            return {"id": obj.id, "name": obj.name}

        d.update(obj.__dict__)
        return d

dann in deiner Ansichtsfunktion

parents = Parent.query.all()
dat = json.dumps({"data": parents}, cls=ParentEncoder)
resp = Response(response=dat, status=200, mimetype="application/json")
return (resp)

es funktioniert gut, obwohl die Eltern Beziehungen haben