¿Cómo serializar el resultado de SqlAlchemy a JSON?

Django tiene una buena serialización automática de modelos ORM devueltos desde el formato DB a JSON.

¿Cómo serializar el resultado de la consulta SQLAlchemy al formato JSON?

Intenté jsonpickle.encode pero codifica el objeto de consulta en sí. Intenté json.dumps(items) pero devuelve

 TypeError:  is not JSON serializable 

¿Es realmente tan difícil serializar objetos ORM de SQLAlchemy a JSON / XML? ¿No hay un serializador predeterminado para ello? Es una tarea muy común serializar los resultados de la consulta ORM hoy en día.

Lo que necesito es simplemente devolver la representación de datos JSON o XML del resultado de la consulta SQLAlchemy.

El resultado de la consulta de objetos SQLAlchemy en formato JSON / XML es necesario para ser usado en javascript datagird (JQGrid http://www.trirand.com/blog/ )

Una implementación plana

Podrías usar algo como esto:

 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) 

y luego convertir a JSON usando:

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

Ignorará los campos que no son codificables (configúrelos en ‘Ninguno’).

No amplía las relaciones de forma automática (ya que esto podría llevar a autoreferencias y hacer un bucle para siempre).

Una implementación recursiva, no circular.

Sin embargo, si prefieres bucear para siempre, puedes usar:

 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 

Y luego codificar objetos usando:

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

Esto codificaría a todos los niños, a todos sus hijos y a todos sus hijos … Codifique potencialmente toda su base de datos, básicamente. Cuando alcanza algo que está codificado anteriormente, lo codificará como ‘Ninguno’.

Una aplicación recursiva, posiblemente circular, selectiva.

Otra alternativa, probablemente mejor, es poder especificar los campos que desea expandir:

 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 

Ahora puedes llamarlo con:

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

Para expandir solo los campos de SQLAlchemy llamados ‘padres’, por ejemplo.

Usted podría simplemente dar salida a su objeto como un dict:

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

Y luego usas User.as_dict () para serializar tu objeto.

Como se explica en Convertir el objeto fila sqlalchemy a dict de python

Puedes convertir un RowProxy a un dict como este:

  d = dict(row.items()) 

Luego, serialice eso a JSON (tendrá que especificar un codificador para cosas como valores de datetime ) No es tan difícil si solo desea un registro (y no una jerarquía completa de registros relacionados).

 json.dumps([(dict(row.items())) for row in rs]) 

Recomiendo usar un malvavisco de biblioteca reciente. Le permite crear serializadores para representar sus instancias de modelo con soporte para relaciones y objetos nesteds.

Echa un vistazo a theier SQLAlchemy Example .

El paquete Flask-JsonTools tiene una implementación de la clase Base JsonSerializableBase para sus modelos.

Uso:

 from sqlalchemy.ext.declarative import declarative_base from flask.ext.jsontools import JsonSerializableBase Base = declarative_base(cls=(JsonSerializableBase,)) class User(Base): #... 

Ahora el modelo de User es mágicamente serializable.

Si su marco no es Flask, simplemente puede agarrar el código

Por razones de seguridad, nunca debe devolver todos los campos del modelo. Prefiero elegirlos selectivamente.

La encoding json de Flask ahora admite UUID, datetime y relaciones (y se agregaron query y query_class para la clase db.Model de db.Model ). He actualizado el codificador de la siguiente manera:

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 

Con esto, opcionalmente puedo agregar una propiedad __json__ que devuelve la lista de campos que deseo codificar:

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

Agrego @jsonapi a mi vista, devuelvo la lista de resultados y luego mi salida es la siguiente:

 [ { "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" } ] 

Puedes usar la introspección de SqlAlchemy como esto:

 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/',methods=['GET']) def getContact(id): contact = Contacts.query.get(id) return jsonify(contact.toDict()) 

Inspírate con una respuesta aquí: Convierte el objeto sqlalchemy row a dict de Python

Una explicación más detallada. En su modelo, agregue:

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

El str() es para python 3, por lo que si usa python 2 use unicode() . Debe ayudar a deserializar fechas. Puede eliminarlo si no se trata con ellos.

Ahora puede consultar la base de datos de esta manera

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

First() se necesita First() para evitar errores extraños. as_dict() ahora deserializará el resultado. Después de la deserialización, está listo para ser convertido a json.

 jsonify(some_result) 

No es tan sencillo. Escribí un código para hacer esto. Todavía estoy trabajando en ello, y utiliza el marco MochiKit. Básicamente, traduce objetos compuestos entre Python y Javascript utilizando un proxy y conversores JSON registrados.

El lado del navegador para los objetos de la base de datos es db.js. Necesita el origen de proxy Python básico en proxy.js .

En el lado de Python está el módulo proxy de base. Luego, finalmente, el codificador de objetos SqlAlchemy en webserver.py . También depende de los extractores de metadatos que se encuentran en el archivo models.py .

Serialización personalizada y deserialización.

“from_json” (método de clase) construye un objeto Model basado en datos json.

“deserialize” podría llamarse solo en la instancia, y fusionar todos los datos de json en la instancia del Modelo.

“serializar” – serialización recursiva

Se necesita la propiedad __write_only__ para definir propiedades de solo escritura (“password_hash” por ejemplo).

 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, eg 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 

Aquí hay una solución que le permite seleccionar las relaciones que desea incluir en su salida tan profundamente como le gustaría ir. NOTA: Esta es una reescritura completa que toma un dict / str como un argumento en lugar de una lista. arregla algunas cosas ..

 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 

así que para un ejemplo usando persona / familia / hogares / habitaciones … convirtiéndolo en json todo lo que necesita es

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

Si bien la pregunta original se remonta a un tiempo atrás, el número de respuestas aquí (y mis propias experiencias) sugieren que es una pregunta no trivial con muchos enfoques diferentes de complejidad variable con diferentes compromisos.

Es por eso que construí la biblioteca de SQLAthanor que extiende el ORM declarativo de SQLAlchemy con un soporte configurable de serialización / des-serialización que tal vez quiera ver.

La biblioteca soporta:

  • Python 2.7, 3.4, 3.5 y 3.6.
  • Versiones SQLAlchemy 0.9 y superiores
  • serialización / des-serialización a / desde JSON, CSV, YAML, y Python dict
  • serialización / des-serialización de columnas / atributos, relaciones, propiedades híbridas y proxies de asociación
  • habilitación e inhabilitación de la serialización para formatos y columnas / relaciones / atributos particulares (por ejemplo, desea admitir un valor de password entrada , pero nunca incluir uno de salida )
  • Pre-serialización y procesamiento de valor posterior a la deserialización (para validación o coerción de tipo)
  • una syntax bastante sencilla que es a la vez Pythonic y perfectamente compatible con el enfoque propio de SQLAlchemy

Puede consultar los documentos completos (¡espero!) Aquí: https://sqlathanor.readthedocs.io/en/latest

¡Espero que esto ayude!

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

Pensé que jugaría un pequeño código de golf con este.

FYI: Estoy usando automap_base ya que tenemos un esquema diseñado de forma separada de acuerdo con los requisitos del negocio. Acabo de comenzar a usar SQLAlchemy hoy, pero la documentación indica que automap_base es una extensión de declarative_base, que parece ser el paradigma típico en el SQLAlchemy ORM, así que creo que debería funcionar.

No resulta sofisticado con las siguientes claves foráneas según la solución de Tjorriemorrie , pero simplemente hace coincidir las columnas con los valores y maneja los tipos de Python por str () – los valores de las columnas. Nuestros valores son Python datetime.time y decimal. Resultados de tipo de clase decimal para que haga el trabajo.

Espero que esto ayude a los transeúntes!

El siguiente código serializará el resultado de sqlalchemy a 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) 

Llamando a la diversión,

 def all_products(): all_products = Products.query.all() return to_array(all_products) 

El AlchemyEncoder es maravilloso, pero a veces falla con valores decimales. Aquí hay un codificador mejorado que resuelve el problema decimal:

 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) 

Sé que este es un post bastante más viejo. Tomé la solución dada por @SashaB y la modifiqué según mi necesidad.

Le agregué lo siguiente:

  1. Lista de ignorar campos: una lista de campos que se ignorarán al serializar
  2. Lista de reemplazo de campos: un diccionario que contiene nombres de campos que se reemplazarán por valores mientras se serializa.
  3. Métodos eliminados y baseQuery ser serializados

Mi código es el siguiente:

 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 

Espero que ayude a alguien!

Use el serializador incorporado en SQLAlchemy:

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

Si está transfiriendo el objeto entre sesiones, recuerde separar el objeto de la sesión actual usando session.expunge(obj) . Para adjuntarlo de nuevo, solo session.add(obj) .

Bajo Frasco, esto funciona y maneja los campos de fecha y hora, transformando un campo de tipo.
'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) 

Los chokes de serializador integrados con utf-8 no pueden decodificar un byte de inicio no válido para algunas entradas. En su lugar, fui con:

 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 

Mi toma utilizando (¿demasiados?) Diccionarios:

 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 

Ejecutar con matraz (incluido jsonify) y flask_sqlalchemy para imprimir salidas como JSON.

Llame a la función con jsonify (serialize ()).

Funciona con todas las consultas de SQLAlchemy que he intentado hasta ahora (ejecutando SQLite3)