Comment sérialiser le résultat SqlAlchemy à JSON?

Django a une bonne sérialisation automatique des modèles ORM retournés du format DB au format JSON.

comment sérialiser le résultat de requête SQLAlchemy au format JSON?

j'ai essayé jsonpickle.encode mais il Code l'objet de requête lui-même. J'ai essayé json.dumps(items) mais il retourne

TypeError: <Product('3', 'some name', 'some desc')> is not JSON serializable

est-il vraiment si difficile de sérialiser des objets SQLAlchemy ORM à JSON /XML? N'est-ce pas là un défaut sérialiseur? C'est une tâche très commune de serialize ORM résultats de requêtes de nos jours.

ce dont j'ai besoin est juste de retourner la représentation de données JSON ou XML du résultat de requête SQLAlchemy.

SQLAlchemy objects query result in JSON / XML format is needed to be used in javascript DataGrid (JQGrid http://www.trirand.com/blog / )

117
demandé sur codegeek 2011-02-17 00:04:05

18 réponses

Un plat de mise en œuvre

vous pourriez utiliser quelque chose comme ceci:

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)

et ensuite convertir en JSON en utilisant:

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

il ignorera les champs qui ne sont pas codables (mettez-les à 'None').

il ne s'étend pas automatiquement les relations (puisque cela pourrait conduire à des références personnelles, et boucle pour toujours).

récursive, non-circulaire de mise en œuvre

Si, cependant, vous préférez boucler pour toujours, vous pouvez utiliser:

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

et ensuite encoder des objets en utilisant:

print json.dumps(e, cls=new_alchemy_encoder(), check_circular=False)

Cela permettrait d'encoder tous les enfants, et tous leurs enfants, et tous leurs enfants... Potentiellement encoder toute votre base de données, en gros. Quand il atteint quelque chose de son encodé avant, il l'encodera comme 'None'.

récursive, peut-être-circulaire, la mise en œuvre sélective

un autre une alternative, probablement meilleure, est de pouvoir spécifier les champs que vous voulez étendre:

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

vous pouvez maintenant l'appeler avec:

print json.dumps(e, cls=new_alchemy_encoder(False, ['parents']), check_circular=False)

pour étendre seulement les champs de SQLAlchemy appelé 'parents', par exemple.

95
répondu Sasha B 2012-05-19 10:45:19

vous pourriez juste sortir votre objet comme un dict:

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

et ensuite vous utilisez User.as_dict () pour sérialiser votre objet.

comme expliqué dans convertir sqlalchemy row objet en Python dict

190
répondu charlax 2017-05-23 12:34:41

vous pouvez convertir un RowProxy en dict comme ceci:

 d = dict(row.items())

alors sérialisez cela à JSON (vous devrez spécifier un encodeur pour des choses comme datetime valeurs ) Il n'est pas difficile si vous voulez juste un enregistrement ( et non pas une hiérarchie complète des documents connexes ).

json.dumps([(dict(row.items())) for row in rs])
41
répondu Nick Perkins 2016-06-12 00:04:30

je recommande l'utilisation d'une bibliothèque récente en surface marshmallow . Il vous permet de créer des sérialiseurs pour représenter vos instances de modèle avec un support aux relations et aux objets imbriqués.

Avoir un coup d'oeil à theier SQLAlchemy Exemple .

28
répondu Yasir Hantoush 2016-05-12 18:40:01

en Flacon de JsonTools paquet a une mise en œuvre de JsonSerializableBase de la classe de Base pour vos modèles.

Utilisation:

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

Base = declarative_base(cls=(JsonSerializableBase,))

class User(Base):
    #...

maintenant le modèle User est magiquement sérialisable.

si votre cadre n'est pas flasque, vous pouvez simplement saisir le code

14
répondu kolypto 2014-09-07 23:46:49

pour des raisons de sécurité, vous ne devez jamais retourner tous les champs du modèle. Je préfère choisir de façon sélective.

L'encodage json du flacon de

supporte maintenant les UUID, datetime et les relations (et a ajouté query et query_class pour la classe flask_sqlalchemy db.Model ). J'ai mis à jour l'encodeur comme suit:

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

avec ceci je peux éventuellement ajouter une propriété __json__ qui renvoie la liste des champs que je souhaite encoder:

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

j'ajoute @jsonapi à ma vue, renvoie la liste des résultats et ensuite ma sortie est la suivante:

[

{

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

]
11
répondu Tjorriemorrie 2015-07-23 13:34:07

vous pouvez utiliser introspection de SqlAlchemy comme ceci:

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

s'inspirer d'une réponse ici : Convertir sqlalchemy ligne d'objet de python dict

7
répondu phico 2017-09-12 15:54:14

Il n'est pas si compliqué. J'ai écrit un code pour faire ça. Je travaille encore dessus,et il utilise le cadre MochiKit. Il traduit essentiellement des objets composés entre Python et Javascript en utilisant un proxy et des convertisseurs JSON enregistrés.

Navigateur de côté pour les objets de base de données est db.js Il a besoin de la source de base du proxy Python dans proxy.js .

du côté Python il y a la base module de proxy . Puis finalement L'encodeur D'objet SqlAlchemy dans webserver.py . Il dépend également des extracteurs de métadonnées trouvés dans le models.py fichier.

3
répondu Keith 2011-02-16 22:23:16

Personnaliser la sérialisation et la désérialisation.

"from_json" (méthode de classe) construit un objet modèle basé sur les données json.

"deserialize" ne peut être appelé qu'en instance, et fusionne toutes les données de json en instance de modèle.

"serialize" - serialization récursive

__écrire_seulement__ biens est nécessaire pour définir les propriétés d'écriture seulement ("password_hash" par exemple).

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
2
répondu Артем Федоров 2016-04-04 15:16:22

Voici une solution qui vous permet de sélectionner les relations que vous souhaitez inclure dans votre sortie aussi profond que vous voulez aller. NOTE: il s'agit d'une réécriture complète prenant un dict/str comme arg plutôt qu'une liste. corrige quelques trucs..

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

donc pour un exemple utilisant personne/famille/maisons/chambres... transformer en json tout ce dont vous avez besoin est

json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))
2
répondu tahoe 2016-12-19 19:48:45

alors que la question originale remonte à un certain temps, le nombre de réponses ici (et mes propres expériences) suggèrent qu'il s'agit d'une question non triviale avec beaucoup d'approches différentes de complexité variable avec des compromis différents.

C'est pourquoi j'ai construit la bibliothèque SQLAthanor qui étend L'ORM déclaratif de SQLAlchemy avec le soutien configurable serialization/de-serialization que vous pourriez vouloir jeter un oeil.

le supports de bibliothèque:

  • Python 2.7, 3.4, 3.5 et 3.6.
  • SQLAlchemy versions 0.9 et plus
  • sérialisation/dé-sérialisation/à partir de JSON, CSV, YAML, et Python dict
  • sérialisation/dé-sérialisation de colonnes/attributs, des relations, des hybrides, des propriétés, et de l'association des procurations
  • activer et désactiver la sérialisation pour des formats particuliers et colonnes/relations/attributs (par exemple, vous souhaitez soutenir une entrant password de la valeur, mais jamais un sortant )
  • pré-sérialisation et post-désérialisation du traitement des valeurs (pour la validation ou de contrainte de type)
  • une syntaxe assez simple qui est à la fois pythonique et parfaitement compatible avec la propre approche de SQLAlchemy

vous pouvez vérifier l' (je l'espère!) documents complets ici: https://sqlathanor.readthedocs.io/en/latest

Espérons que cette aide!

2
répondu Chris Modzelewski 2018-07-22 20:58:44
def alc2json(row):
    return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])

j'ai pensé jouer au golf avec celui-ci.

FYI: j'utilise automap_base puisque nous avons un schéma conçu séparément en fonction des besoins de l'entreprise. Je viens de commencer à utiliser SQLAlchemy aujourd'hui, mais la documentation stipule que automap_base est une extension à declarative_base qui semble être le paradigme typique dans L'ORM SQLAlchemy donc je crois que cela devrait fonctionner.

il ne obtenez fantaisie avec les clés étrangères suivantes par tjorriemorrie 's solution, mais il correspond simplement les colonnes aux valeurs et gère les types Python par str()-ing les valeurs de colonne. Nos valeurs sont Python datetime.le temps et les décimales.Classe Decimal type de résultats, il fait le travail.

espérons que cela aide les passants!

1
répondu hpatel71 2017-05-23 12:26:20

Une explication plus détaillée. Dans votre modèle, Ajouter:

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

le str() est pour python 3 donc si vous utilisez python 2 Utilisez unicode() . Ça devrait aider à désérialiser les dates. Vous pouvez le supprimer si pas de traiter avec ces.

vous pouvez maintenant interroger la base de données comme ceci

some_result = User.query.filter_by(id=current_user.id).first().as_dict()

First() est nécessaire pour éviter les erreurs étranges. as_dict() désérialiser le résultat. Après la désérialisation, il est prêt à tourner à json

jsonify(some_result)
1
répondu Patrick Mutuku 2018-05-11 19:29:34

le code suivant sérialisera le résultat de sqlalchemy à 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) 

l'Appel du plaisir,

def all_products():
    all_products = Products.query.all()
    return to_array(all_products)
1
répondu Chirag Vora 2018-06-13 12:02:37

je sais que c'est tout à fait un vieux poste. J'ai pris la solution donnée par @SashaB et modifié selon mon besoin.

j'y ai ajouté les choses suivantes:

    "151960920 Champ" ignorer: Une liste de champs à être ignoré lors de la sérialisation
  1. zone Remplacer Liste: un dictionnaire contenant des noms de zones à remplacer par des valeurs lors de la sérialisation.
  2. méthodes enlevées et BaseQuery obtenir sérialisé

mon code est le suivant:

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

Espère que cela aide quelqu'un!

0
répondu srahul07 2016-05-14 07:25:58

utiliser le serializer intégré dans SQLAlchemy:

from sqlalchemy.ext.serializer import loads, dumps
obj = MyAlchemyObject()
# serialize object
serialized_obj = dumps(obj)

# deserialize object
obj = loads(serialized_obj)

si vous transférez l'objet entre les sessions, n'oubliez pas de détacher l'objet de la session en cours en utilisant session.expunge(obj) . Pour l'attacher à nouveau, il suffit de faire session.add(obj) .

0
répondu chribsen 2016-12-13 11:31:06

sous flasque, cela fonctionne et traite les champs de données, transformant un champ de type

'time': datetime.datetime(2018, 3, 22, 15, 40) en

"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)
0
répondu belvederef 2018-03-10 22:42:47

Mon prendre en utilisant (trop?) dictionnaires:

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

tournant avec flask (y compris jsonify) et flask_sqlalchemy pour imprimer des sorties comme JSON.

appelle la fonction avec jsonify(serialize()).

fonctionne avec toutes les requêtes SQLAlchemy que j'ai essayé jusqu'à présent (exécution SQLite3)

-2
répondu Sawyer Brooks 2016-06-08 15:17:15