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

192

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

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

Lo intenté jsonpickle.encodepero codifica el objeto de consulta en sí. Lo intenté json.dumps(items)pero vuelve

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

¿Es realmente tan difícil serializar objetos SQLAlchemy ORM a JSON / XML? ¿No hay ningún serializador predeterminado para ello? Hoy en día es una tarea muy común serializar los resultados de las consultas ORM.

Lo que necesito es 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 utilizado en JavaScript DataGird (JQGrid http://www.trirand.com/blog/ )

Zelid
fuente
Esta es una solución que me funciona. ingrese la descripción del enlace aquí
octaedro

Respuestas:

129

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 expande automáticamente las relaciones (ya que esto podría conducir a auto-referencias y bucles para siempre).

Una implementación recursiva, no circular

Sin embargo, si prefieres un bucle 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 hijos, y a todos sus hijos, y a todos sus hijos ... Básicamente, codifique potencialmente toda su base de datos. Cuando alcanza algo que está codificado anteriormente, lo codificará como 'Ninguno'.

Una implementació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 puede 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.

Sasha B
fuente
esa es una gran respuesta, sin embargo, obtengo un "no se puede codificar" BaseQuery "cada vez que se relaciona con los métodos no planos, ¿alguna idea?
Ben Kilah
1
@SashaB ¿Qué tal si se enfoca de manera más granular contra los casos en los que se repite una relación? Por ejemplo, si tengo online_ordery address, ambos con una relación con user, pero online_ordertambién tiene una relación con address. Si quisiera serializar todo esto, tendría que incluirlo addressen el fields_to_expand, pero no quisiera serializar de forma redundante addressdebido a su relación con ambos usery online_order.
Chrispy
2
@BenKilah Déjame adivinar, estás usando Flask-SqlAlchemy y tus modelos heredan de db.Model, no Base. Si ese es el caso, modifique for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata']:para que se lea for field in [x for x in dir(obj) if not x.startswith('_') and x != 'metadata' and not x.startswith('query')]:. Tenga en cuenta que esta solución le impedirá tener una propiedad / relación con el nombre 'query'
Pakman
igual que yo, pero mucho más complejo. stackoverflow.com/questions/7102754/…
tyan
2
Puede usar mi solución github.com/n0nSmoker/SQLAlchemy-serializer
n0nSmoker
273

Simplemente puede generar su objeto como un diccionario:

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 objeto de fila sqlalchemy a python dict

charlax
fuente
2
@charlax, ¿cómo solucioné una fecha y hora? Al usar esto obtengo 'datetime.datetime (2013, 3, 22, 16, 50, 11) no es serializable JSON' cuando hago json.dumps
Asken
1
Es responsabilidad del JSONEncoderobjeto. Puede subclasificarlo para definir su propio codificador para algún objeto, incluida la fecha y hora. Tenga en cuenta que Flask, por ejemplo, admite la codificación de fecha y hora en JSON fuera de la caja (con la última versión).
charlax
3
Si usa el método "declarativo" de sqlalchemy, puede agregar algo como esto a una clase Base personalizada; esto es bastante útil ya que puede llamar a my_orm_object.toDict () en cualquier objeto ORM que tenga. Del mismo modo, puede definir un método .toJSON () que utiliza su método toDict y un codificador personalizado para manejar fechas, blobs, etc.
FredL
77
para admitir también fecha y hora:return {c.name: unicode(getattr(self, c.name)) for c in self.__table__.columns}
Shoham
1
Esto no funciona si sus variables de clase no son las mismas que los nombres de sus columnas. ¿Alguna idea de cómo obtener los nombres de clase en su lugar?
James Burke
55

Puede 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 datetimevalores) 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])
Nick Perkins
fuente
1
Esto funciona para mi consulta sql personalizada con db.engine.connect () como con: rs = con.execute (sql)
JZ.
1
Esto es mucho más simple y funciona. ¿Cuál es la diferencia entre esta respuesta y la respuesta aceptada?
Sundeep
46

Recomiendo usar malvavisco . Le permite crear serializadores para representar sus instancias de modelo con soporte para relaciones y objetos anidados.

Aquí hay un ejemplo truncado de sus documentos. Tome el modelo ORM Author:

class Author(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    first = db.Column(db.String(80))
    last = db.Column(db.String(80))

Un esquema de malvavisco para esa clase se construye así:

class AuthorSchema(Schema):
    id = fields.Int(dump_only=True)
    first = fields.Str()
    last = fields.Str()
    formatted_name = fields.Method("format_name", dump_only=True)

    def format_name(self, author):
        return "{}, {}".format(author.last, author.first)

... y usado así:

author_schema = AuthorSchema()
author_schema.dump(Author.query.first())

... produciría una salida como esta:

{
        "first": "Tim",
        "formatted_name": "Peters, Tim",
        "id": 1,
        "last": "Peters"
}

Eche un vistazo a su ejemplo completo de Flask-SQLAlchemy .

Una biblioteca llamada marshmallow-sqlalchemyespecíficamente integra SQLAlchemy y malvavisco. En esa biblioteca, el esquema para el Authormodelo descrito anteriormente se ve así:

class AuthorSchema(ModelSchema):
    class Meta:
        model = Author

La integración permite inferir los tipos de campo a partir de los Columntipos SQLAlchemy .

malvavisco-sqlalchemy aquí.

Yasir Hantoush
fuente
12
También encontré marshmallow-sqlalchemy.readthedocs.io/en/latest que simplifica la generación del esquema
Foo L
46

Python 3.7+ y Flask 1.1+ pueden usar el paquete de clases de datos incorporado

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 __name__ == "__main__":
  users = User(email="[email protected]"), User(email="[email protected]")
  db.create_all()
  db.session.add_all(users)
  db.session.commit()
  app.run()

La /users/ruta ahora devolverá una lista de usuarios.

[
  {"email": "[email protected]", "id": 1},
  {"email": "[email protected]", "id": 2}
]

Auto-serializar modelos relacionados

@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

La respuesta de jsonify(account)sería esta.

{  
   "id":1,
   "users":[  
      {  
         "email":"[email protected]",
         "id":1
      },
      {  
         "email":"[email protected]",
         "id":2
      }
   ]
}

Sobrescribe el codificador JSON predeterminado

from flask.json import JSONEncoder


class CustomJSONEncoder(JSONEncoder):
  "Add support for serializing timedeltas"

  def default(o):
    if type(o) == datetime.timedelta:
      return str(o)
    elif type(o) == datetime.datetime:
      return o.isoformat()
    else:
      return super().default(o)

app.json_encoder = CustomJSONEncoder      
tom
fuente
1
Esto parece el tipo correcto de simple. ¿Funciona también para la deserialización?
Ender2050
Puede convertir un diccionario de JSON analizado en un modelo utilizando el desempaquetado de argumentos de palabras clave:data = request.json['user']; user = User(**data)
tom
3
Tenga en cuenta que id: int = Columnfuncionará, pero id = Columnno lo hará, parece que DEBE declarar la escritura estática para que el json serialice el campo, de lo contrario obtendrá un {}objeto vacío .
Ambroise Rabier
1
Esto funcionó para mí, ¿por qué no es esta la respuesta aceptada? He estado jugando alrededor de app_context durante horas para que esto funcione con Flask-Marshmallow.
Nick Dat Le
1
Me funcionó a mi también. Tenga en cuenta que si estás en Python 3.6, usted querrá sólo tiene que instalar el paquete: pipenv install dataclasses. Y luego funcionará bien.
AleksandrH
14

El paquete Flask-JsonTools tiene una implementación de la clase JsonSerializableBase Base 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 Usermodelo es mágicamente serializable.

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

Kolypto
fuente
2
Esto solo resuelve la mitad del problema, ya que solo serializa una sola fila. ¿Cómo serializar todo el resultado de la consulta?
Steve Bennett
@SteveBennett usa el jsonapi de jsontools para codificar la respuesta. Eso codificará automáticamente el objeto
devuelto
Tengo un modelo de sqlalchemy muy simple, y obtengo: TypeError: <ORM.State object at 0x03577A50> no es serializable JSON
Matej
1
Finalmente funcionó llamando explícitamente a __json __ () en mi objeto modelo: return my_object .__ json __ ()
Matej
La biblioteca no funciona con Flask 1.0 y superior, import flask.ext.whateverya que ya no es compatible con Flask 1.0.
Adarsh ​​Madrecha
14

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

La codificación json de Flask ahora admite UUID, fecha y hora y relaciones (y agregado queryy query_classpara la db.Modelclase flask_sqlalchemy ). 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 __json__propiedad 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"
}

]
Tjorriemorrie
fuente
¡Hermoso! Una vez más, pruebe que a veces no necesita un paquete gordo para cada pequeña tarea estúpida: que aprender DSL puede ser más difícil que hacerlo de la manera "difícil". Miré muchos paquetes JSON y REST antes de aterrizar aquí. Es cierto que esto todavía requiere un paquete, flask_jsontools (a añadir @jsonapia @app.routeen views.py etc), pero me encanta la simplicidad de la misma. Creo que es barato Flask agregó datetime pero no date, así que lo agregué yo mismo a json_encoder.py : value=...^ if isinstance(value, date):^ data[field] = datetime.combine(value, time.min).isoformat()^ else:^try:...
juanitogan
10

Puede usar la introspección de SqlAlchemy como esta:

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

Inspírate con una respuesta aquí: Convierte el objeto de fila sqlalchemy a python dict

phico
fuente
5

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 el uso de pitón 2 uso unicode(). Debería ayudar a deserializar las fechas. Puede eliminarlo si no se trata de eso.

Ahora puede consultar la base de datos de esta manera

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

First()Es necesario para evitar errores extraños. as_dict()ahora deserializará el resultado. Después de la deserialización, está listo para convertirse en json

jsonify(some_result)
Patrick Mutuku
fuente
3

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

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

En el lado de Python está el módulo proxy 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 .

Keith
fuente
Bastante complicado a primera vista ... Lo que necesito es obtener el resultado de la consulta de objetos SQLAlchemy en formato JSON / XML para usarlo en javascript datagird (JQGrid trirand.com/blog )
Zelid
A veces, los problemas son más complicados de lo que se exponen a primera vista ... Esto maneja los objetos devueltos como claves foráneas e intenta evitar la recursión infinita que ocurre con las relaciones profundamente anidadas. Sin embargo, probablemente podría escribir algunas consultas personalizadas que solo devuelvan tipos base y serialicen aquellas con simplejson directamente.
Keith
1
Correcto, tal vez realmente iré a consultar dictos usando SQLAlchemy y usaré los beneficios de ORM que realiza solo acciones de guardar / actualizar.
Zelid
3

Si bien la pregunta original se remonta un tiempo, la cantidad de respuestas aquí (y mis propias experiencias) sugieren que es una pregunta no trivial con muchos enfoques diferentes de complejidad variable con diferentes compensaciones.

Por eso construí el SQLAthanor biblioteca que extiende el ORM declarativo de SQLAlchemy con soporte de serialización / configurable que es posible que desee echar un vistazo.

La biblioteca admite:

  • Python 2.7, 3.4, 3.5 y 3.6.
  • SQLAlchemy versiones 0.9 y superiores
  • serialización / deserialización a / desde JSON, CSV, YAML y Python dict
  • serialización / deserialización de columnas / atributos, relaciones, propiedades híbridas y proxies de asociación
  • Habilitación e inhabilitación de la serialización para formatos particulares y columnas / relaciones / atributos (por ejemplo, si desea admitir un valor entrante password , pero nunca incluir uno saliente )
  • procesamiento de valor previo a la serialización y posterior a la deserialización (para validación o coerción de tipo)
  • una sintaxis bastante sencilla que es a la vez pitónica y perfectamente coherente con el enfoque de SQLAlchemy

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

¡Espero que esto ayude!

Chris Modzelewski
fuente
2

Serialización y deserialización personalizadas.

"from_json" (método de clase) crea un objeto Modelo basado en datos json.

"deserialize" podría llamarse solo en la instancia y fusionar todos los datos de json en la instancia de Model.

"serializar" - serialización recursiva

La propiedad __write_only__ es necesaria 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, 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
Артем Федоров
fuente
2

Aquí hay una solución que le permite seleccionar las relaciones que desea incluir en su salida tan profundo como le gustaría ir. NOTA: Esta es una reescritura completa tomando 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 necesitas es

json.dumps(person.deep_dict(relations={'family':{'homes':'rooms'}}))
Tahoe
fuente
Esto está bien, creo que ponerlo en su clase base para que todos los objetos lo tengan. Te dejaré la codificación json ...
tahoe
Tenga en cuenta que esta versión recibirá todas las relaciones de la lista así que tenga cuidado que proporciona las relaciones con un montón de artículos ...
Tahoe
1
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 por separado de acuerdo con los requisitos comerciales. 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 SQLAlchemy ORM, así que creo que esto debería funcionar.

No es elegante seguir las claves foráneas según la solución de Tjorriemorrie , sino que simplemente hace coincidir las columnas con los valores y maneja los tipos de Python mediante str (): los valores de las columnas. Nuestros valores consisten en Python datetime.time y decimal. Resultados de tipo de clase decimal para que haga el trabajo.

¡Espero que esto ayude a los transeúntes!

hpatel71
fuente
1

Sé que esta es una publicación bastante antigua. Tomé la solución dada por @SashaB y modifiqué según mi necesidad.

Le agregué las siguientes cosas:

  1. Lista de ignorar campos: una lista de campos a ignorar durante la serialización
  2. Lista de reemplazo de campo: un diccionario que contiene nombres de campo para ser reemplazados por valores durante la serialización.
  3. Métodos eliminados y BaseQuery que se serializa

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!

srahul07
fuente
1

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 nuevamente, solo hazlo session.add(obj).

chribsen
fuente
Ingenioso, pero no se convierte a JSON.
blakev
2
Para la 'serialización' de JSON, consulte marshmallow-sqlalchemy . Definitivamente la mejor solución cuando expones objetos a clientes. marshmallow-sqlalchemy.readthedocs.io
chribsen
El módulo serializador solo es apropiado para estructuras de consulta. No es necesario para: instancias de clases definidas por el usuario. Estos no contienen referencias a motores, sesiones o construcciones de expresión en el caso típico y se pueden serializar directamente.
thomasd
1

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 diversión

def all_products():
    all_products = Products.query.all()
    return to_array(all_products)
Chirag Vora
fuente
1

El AlchemyEncoder es maravilloso, pero a veces falla con los 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)
agarwal de vinagre
fuente
1

Cuando se usa sqlalchemy para conectarse a un db I, esta es una solución simple que es altamente configurable. Usa pandas.

import pandas as pd
import sqlalchemy

#sqlalchemy engine configuration
engine = sqlalchemy.create_engine....

def my_function():
  #read in from sql directly into a pandas dataframe
  #check the pandas documentation for additional config options
  sql_DF = pd.read_sql_table("table_name", con=engine)

  # "orient" is optional here but allows you to specify the json formatting you require
  sql_json = sql_DF.to_json(orient="index")

  return sql_json
chandler9
fuente
1
step1:
class CNAME:
   ...
   def as_dict(self):
       return {item.name: getattr(self, item.name) for item in self.__table__.columns}

step2:
list = []
for data in session.query(CNAME).all():
    list.append(data.as_dict())

step3:
return jsonify(list)
邢 烽 朔
fuente
3
Los volcados de código sin ninguna explicación rara vez son útiles. Stack Overflow se trata de aprender, no de proporcionar fragmentos para copiar y pegar a ciegas. Por favor, editar su pregunta y explicar cómo funciona mejor que lo que el PO siempre.
Chris
0

Bajo Flask, esto funciona y maneja campos de datos de tiempo, 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)
belvederef
fuente
0

Los choques de serializador integrados con utf-8 no pueden decodificar el byte de inicio no válido para algunas entradas. En cambio, 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
RobotHumanos
fuente
0

Tal vez puedas usar una clase como esta

from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy import Table


class Custom:
    """Some custom logic here!"""

    __table__: Table  # def for mypy

    @declared_attr
    def __tablename__(cls):  # pylint: disable=no-self-argument
        return cls.__name__  # pylint: disable= no-member

    def to_dict(self) -> Dict[str, Any]:
        """Serializes only column data."""
        return {c.name: getattr(self, c.name) for c in self.__table__.columns}

Base = declarative_base(cls=Custom)

class MyOwnTable(Base):
    #COLUMNS!

Con eso todos los objetos tienen el to_dictmétodo

Andrex
fuente
0

Mientras usaba algunos objetos sql sin procesar y no definidos, cursor.descriptionparecía que usaba para obtener lo que estaba buscando:

with connection.cursor() as cur:
    print(query)
    cur.execute(query)
    for item in cur.fetchall():
        row = {column.name: item[i] for i, column in enumerate(cur.description)}
        print(row)
jmunsch
fuente
-2

Mi opinión utilizando diccionarios (¿demasiados?):

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

Se ejecuta con flask (incluido jsonify) y flask_sqlalchemy para imprimir salidas como JSON.

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

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

Sawyer Brooks
fuente