Comment sérialiser SqlAlchemy résultat JSON?

Un plat de mise en œuvre de

Vous pouvez 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 de les convertir en JSON à l'aide de:

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

Il ignore les champs qui ne sont pas encodable (mis à 'None').

Il n'a pas d'auto-développer les relations (car cela pourrait l'amener à l'auto-références, et boucle pour toujours).

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

Si, toutefois, vous préférez faire une boucle 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

Puis coder les objets à l'aide de:

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

Ce serait de coder tous les enfants et tous leurs enfants, et tous leurs enfants... Potentiellement coder l'ensemble de votre base de données, dans le fond. Lorsqu'il atteint quelque chose de son encodé en avant, il va encoder comme "Aucun".

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

Une autre solution, sans doute mieux, c'est d'être capable de spécifier les champs que vous souhaitez développer:

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 appeler avec:

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

À seulement étendre SQLAlchemy champs appelés "parents", par exemple.

Vous pouvez convertir un RowProxy à une dict comme ceci:

 d = dict(row.items())

Puis sérialiser que JSON ( vous devrez spécifier un codeur 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])

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 la User modèle est magiquement serializable.

Si votre cadre n'est pas le Flacon, vous pouvez simplement saisir le code

Vous pouvez utiliser l'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 de réponse ici :
Convertir sqlalchemy ligne d'objet de python dict

Il n'est pas si compliqué. J'ai écrit du code pour ce faire. Je travaille toujours sur elle, et il utilise le MochiKit cadre. Essentiellement, il se traduit par des objets composés entre Python et Javascript à l'aide d'un proxy et enregistré JSON convertisseurs.

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

Sur le Python côté, il y a la base module de proxy.
Puis, enfin, la SqlAlchemy objet de l'encodeur dans webserver.py.
Elle dépend aussi de métadonnées extracteurs trouvé dans les models.py fichier.

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: Ceci est une réécriture complète de prendre un dict/str comme un 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 d'utilisation de la personne/famille/maisons/chambres... transformant en json tous vous avez besoin est

json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))

def alc2json(row):
return dict([(col, str(getattr(row,col))) for col in row.__table__.columns.keys()])

Je pensais jouer un peu de code de golf avec celui-ci.

Pour info: je suis en utilisant automap_base depuis que nous avons conçus séparément schéma en fonction des exigences de l'entreprise. J'ai juste commencé à l'aide de SQLAlchemy aujourd'hui, mais la documentation indique que les automap_base est une extension pour declarative_base qui semble être typique de paradigme dans la SQLAlchemy ORM je crois que cela devrait fonctionner.

Il n'a pas obtenir la fantaisie avec des clés étrangères par Tjorriemorrie's solution, mais il a simplement fait correspondre les colonnes de valeurs et de poignées de Python de type str()-ing les valeurs de la colonne. Nos valeurs sont Python datetime.le temps et les décimales.Classe Decimal type de résultats, il fait le travail.

Espère que cela aide à tous les passants!

La AlchemyEncoder est merveilleux, mais échoue parfois avec des valeurs Décimales. Voici une amélioration de l'encodeur qui résout le décimal problème -

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)

Utiliser le construit-dans sérialiseur 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 êtes le transfert de l'objet entre les sessions, n'oubliez pas de détacher l'objet à partir de la session en cours à l'aide de session.expunge(obj).
Pour l'attacher à nouveau, il suffit de faire session.add(obj).

Construit dans le sérialiseur étouffe avec de l'utf-8 ne peut pas décoder invalide octet de départ pour certains intrants. Au lieu de cela, je suis allé avec:

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