Convertir objeto de fila sqlalchemy a python dict

¿Hay una manera simple de iterar sobre pares de nombre y valor de columna?

Mi versión de sqlalchemy es 0.5.6

Aquí está el código de muestra donde intenté usar dict (fila), pero arroja una excepción, TypeError: el objeto 'Usuario' no es iterable

import sqlalchemy
from sqlalchemy import *
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

print "sqlalchemy version:",sqlalchemy.__version__ 

engine = create_engine('sqlite:///:memory:', echo=False)
metadata = MetaData()
users_table = Table('users', metadata,
     Column('id', Integer, primary_key=True),
     Column('name', String),
)
metadata.create_all(engine) 

class User(declarative_base()):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)

    def __init__(self, name):
        self.name = name

Session = sessionmaker(bind=engine)
session = Session()

user1 = User("anurag")
session.add(user1)
session.commit()

# uncommenting next line throws exception 'TypeError: 'User' object is not iterable'
#print dict(user1)
# this one also throws 'TypeError: 'User' object is not iterable'
for u in session.query(User).all():
    print dict(u)

Ejecutar este código en las salidas de mi sistema:

sqlalchemy version: 0.5.6
Traceback (most recent call last):
  File "untitled-1.py", line 37, in <module>
    print dict(u)
TypeError: 'User' object is not iterable

Puede acceder al interno __dict__de un objeto SQLAlchemy, como el siguiente ::

for u in session.query(User).all():
    print u.__dict__

No pude obtener una buena respuesta, así que uso esto:

def row2dict(row):
    d = {}
    for column in row.__table__.columns:
        d[column.name] = str(getattr(row, column.name))

    return d

Editar: si la función anterior es demasiado larga y no es adecuada para algunos gustos, aquí hay una línea única (python 2.7+)

row2dict = lambda r: {c.name: str(getattr(r, c.name)) for c in r.__table__.columns}

Según @zzzeek en los comentarios:

tenga en cuenta que esta es la respuesta correcta para las versiones modernas de SQLAlchemy, suponiendo que "fila" es un objeto de fila central, no una instancia asignada a ORM.

for row in resultproxy:
    row_as_dict = dict(row)

En SQLAlchemy v0.8 y más reciente, use el sistema de inspección .

from sqlalchemy import inspect

def object_as_dict(obj):
    return {c.key: getattr(obj, c.key)
            for c in inspect(obj).mapper.column_attrs}

user = session.query(User).first()

d = object_as_dict(user)

Tenga en cuenta que .keyes el nombre del atributo, que puede ser diferente del nombre de la columna, por ejemplo, en el siguiente caso:

class_ = Column('class', Text)

Este método también funciona para column_property.

las filas tienen una _asdict()función que da un dict

In [8]: r1 = db.session.query(Topic.name).first()

In [9]: r1
Out[9]: (u'blah')

In [10]: r1.name
Out[10]: u'blah'

In [11]: r1._asdict()
Out[11]: {'name': u'blah'}

como @balki mencionó:

El _asdict()método se puede usar si está consultando un campo específico porque se devuelve como KeyedTuple .

In [1]: foo = db.session.query(Topic.name).first()
In [2]: foo._asdict()
Out[2]: {'name': u'blah'}

Mientras que, si no especifica una columna, puede usar uno de los otros métodos propuestos, como el proporcionado por @charlax. Tenga en cuenta que este método solo es válido para 2.7+.

In [1]: foo = db.session.query(Topic).first()
In [2]: {x.name: getattr(foo, x.name) for x in foo.__table__.columns}
Out[2]: {'name': u'blah'}

Antigua pregunta, pero dado que este es el primer resultado para "sqlalchemy row to dict" en Google, merece una mejor respuesta.

El objeto RowProxy que devuelve SqlAlchemy tiene el método items (): http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items

Simplemente devuelve una lista de tuplas (clave, valor). Entonces uno puede convertir una fila a dict usando lo siguiente:

En Python <= 2.6:

rows = conn.execute(query)
list_of_dicts = [dict((key, value) for key, value in row.items()) for row in rows]

En Python> = 2.7:

rows = conn.execute(query)
list_of_dicts = [{key: value for (key, value) in row.items()} for row in rows]

Asumiendo que las siguientes funciones se agregarán a las class Usersiguientes, se devolverán todos los pares clave-valor de todas las columnas:

def columns_to_dict(self):
    dict_ = {}
    for key in self.__mapper__.c.keys():
        dict_[key] = getattr(self, key)
    return dict_

a diferencia de las otras respuestas, se devuelven todos los atributos del objeto, excepto los atributos del Columnnivel de clase del objeto. Por lo tanto, no se incluye _sa_instance_stateningún atributo SQLalchemy o cualquier otro que agregue al objeto. Referencia

EDITAR: Olvídese de decir que esto también funciona en columnas heredadas.

hybrid_propery extensión

Si también desea incluir hybrid_propertyatributos, funcionará lo siguiente:

from sqlalchemy import inspect
from sqlalchemy.ext.hybrid import hybrid_property

def publics_to_dict(self) -> {}:
    dict_ = {}
    for key in self.__mapper__.c.keys():
        if not key.startswith('_'):
            dict_[key] = getattr(self, key)

    for key, prop in inspect(self.__class__).all_orm_descriptors.items():
        if isinstance(prop, hybrid_property):
            dict_[key] = getattr(self, key)
    return dict_

Supongo que aquí marca Columnas con un comienzo _para indicar que desea ocultarlas, ya sea porque accede al atributo mediante un hybrid_propertyo simplemente no desea mostrarlas. Referencia

Tipp all_orm_descriptors también devuelve hybrid_method y AssociationProxy si también desea incluirlos.

Observaciones a otras respuestas

Cada respuesta (como 1 , 2 ) que se basa en el __dict__atributo simplemente devuelve todos los atributos del objeto. Esto podría ser mucho más atributos de lo que deseas. Como triste, esto incluye _sa_instance_stateo cualquier otro atributo que defina en este objeto.

Cada respuesta (como 1 , 2 ) que se basa en la dict()función solo funciona en objetos de fila SQLalchemy devueltos por session.execute()no en las clases con las que define trabajar, como el class Userde la pregunta.

La respuesta de resolución que se basa row.__table__.columnsdefinitivamente no funcionará. row.__table__.columnscontiene los nombres de columna de la base de datos SQL. Estos solo pueden ser iguales al nombre de los atributos del objeto python. Si no, obtienes un AttributeError. Para las respuestas (como 1 , 2 ) basadas en class_mapper(obj.__class__).mapped_table.cesto es lo mismo.

from sqlalchemy.orm import class_mapper

def asdict(obj):
    return dict((col.name, getattr(obj, col.name))
                for col in class_mapper(obj.__class__).mapped_table.c)

Siguiendo la respuesta de @balki, desde SQLAlchemy 0.8 puede usar _asdict (), disponible para objetos KeyedTuple. Esto genera una respuesta bastante directa a la pregunta original. Simplemente, cambie en su ejemplo las dos últimas líneas (el bucle for) para esta:

for u in session.query(User).all():
   print u._asdict()

Esto funciona porque en el código anterior u es un objeto de tipo clase KeyedTuple , ya que .all () devuelve una lista de KeyedTuple. Por lo tanto, tiene el método _asdict () , que devuelve muy bien u como diccionario.

Escribe la respuesta de @STB: AFAIK, y aunque .all () devuelva es una lista de KeypedTuple. Por lo tanto, lo anterior funciona si especifica una columna o no, siempre y cuando esté tratando con el resultado de .all () tal como se aplica a un objeto Query.

La expresión que está iterando evalúa la lista de objetos modelo , no filas. Entonces, el siguiente es el uso correcto de ellos:

for u in session.query(User).all():
    print u.id, u.name

¿Realmente necesitas convertirlos a dictos? Claro, hay muchas maneras, pero entonces no necesitas ORM parte de SQLAlchemy:

result = session.execute(User.__table__.select())
for row in result:
    print dict(row)

Actualización : Echa un vistazo al sqlalchemy.orm.attributesmódulo. Tiene un conjunto de funciones para trabajar con el estado del objeto, que puede ser útil para usted, especialmente instance_dict().

Consulte la respuesta de Alex Brasetvik , puede usar una línea de código para resolver el problema

row_as_dict = [dict(row) for row in resultproxy]

En la sección de comentarios de la Respuesta de Alex Brasetvik, zzzeek, ​​el creador de SQLAlchemy, declaró que este es el "Método correcto" para el problema.

Podrías intentar hacerlo de esta manera.

for u in session.query(User).all():
    print(u._asdict())

Utiliza un método incorporado en el objeto de consulta que devuelve un objeto dictonario del objeto de consulta.

referencias: https://docs.sqlalchemy.org/en/latest/orm/query.html

Encontré esta publicación porque estaba buscando una manera de convertir una fila SQLAlchemy en un dict. Estoy usando SqlSoup ... pero la respuesta fue creada por mí mismo, así que, si pudiera ayudar a alguien, aquí están mis dos centavos:

a = db.execute('select * from acquisizioni_motes')
b = a.fetchall()
c = b[0]

# and now, finally...
dict(zip(c.keys(), c.values()))

Puede convertir el objeto sqlalchemy en un diccionario como este y devolverlo como json / dictionary.

Funciones de ayuda:

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) 

Función del conductor:

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

Dos caminos:

1)

for row in session.execute(session.query(User).statement):
    print(dict(row))

2)

selected_columns = User.__table__.columns
rows = session.query(User).with_entities(*selected_columns).all()
for row in rows :
    print(row._asdict())

Los documentos ofrecen una solución muy simple: ResultRow._asdict()

def to_array(rows):
    return [r._asdict() for r in rows]

def query():
    data = session.query(Table).all()
    return to_array(data)

Así es como lo hace Elixir. El valor de esta solución es que permite incluir recursivamente la representación de las relaciones en el diccionario.

def to_dict(self, deep={}, exclude=[]):
    """Generate a JSON-style nested dict/list structure from an object."""
    col_prop_names = [p.key for p in self.mapper.iterate_properties \
                                  if isinstance(p, ColumnProperty)]
    data = dict([(name, getattr(self, name))
                 for name in col_prop_names if name not in exclude])
    for rname, rdeep in deep.iteritems():
        dbdata = getattr(self, rname)
        #FIXME: use attribute names (ie coltoprop) instead of column names
        fks = self.mapper.get_property(rname).remote_side
        exclude = [c.name for c in fks]
        if dbdata is None:
            data[rname] = None
        elif isinstance(dbdata, list):
            data[rname] = [o.to_dict(rdeep, exclude) for o in dbdata]
        else:
            data[rname] = dbdata.to_dict(rdeep, exclude)
    return data

¡Con este código también puede agregar a su consulta "filtro" o "unirse" y esto funciona!

query = session.query(User)
def query_to_dict(query):
        def _create_dict(r):
            return {c.get('name'): getattr(r, c.get('name')) for c in query.column_descriptions}

    return [_create_dict(r) for r in query]

class User(object):
    def to_dict(self):
        return dict([(k, getattr(self, k)) for k in self.__dict__.keys() if not k.startswith("_")])

Eso debería funcionar.

Tengo una variación sobre la respuesta de Marco Mariani, expresada como decoradora. La principal diferencia es que manejará listas de entidades, además de ignorar de forma segura algunos otros tipos de valores de retorno (lo cual es muy útil al escribir pruebas usando simulacros):

@decorator
def to_dict(f, *args, **kwargs):
  result = f(*args, **kwargs)
  if is_iterable(result) and not is_dict(result):
    return map(asdict, result)

  return asdict(result)

def asdict(obj):
  return dict((col.name, getattr(obj, col.name))
              for col in class_mapper(obj.__class__).mapped_table.c)

def is_dict(obj):
  return isinstance(obj, dict)

def is_iterable(obj):
  return True if getattr(obj, '__iter__', False) else False

Para completar la respuesta de @Anurag Uniyal, aquí hay un método que seguirá recursivamente las relaciones:

from sqlalchemy.inspection import inspect

def to_dict(obj, with_relationships=True):
    d = {}
    for column in obj.__table__.columns:
        if with_relationships and len(column.foreign_keys) > 0:
             # Skip foreign keys
            continue
        d[column.name] = getattr(obj, column.name)

    if with_relationships:
        for relationship in inspect(type(obj)).relationships:
            val = getattr(obj, relationship.key)
            d[relationship.key] = to_dict(val) if val else None
    return d

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    first_name = Column(TEXT)
    address_id = Column(Integer, ForeignKey('addresses.id')
    address = relationship('Address')

class Address(Base):
    __tablename__ = 'addresses'
    id = Column(Integer, primary_key=True)
    city = Column(TEXT)


user = User(first_name='Nathan', address=Address(city='Lyon'))
# Add and commit user to session to create ids

to_dict(user)
# {'id': 1, 'first_name': 'Nathan', 'address': {'city': 'Lyon'}}
to_dict(user, with_relationship=False)
# {'id': 1, 'first_name': 'Nathan', 'address_id': 1}

Con python 3.8+, podemos hacer esto con dataclass y el asdictmétodo que viene con él:

from dataclasses import dataclass, asdict

from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy import Column, String, Integer, create_engine

Base = declarative_base()
engine = create_engine('sqlite:///:memory:', echo=False)


@dataclass
class User(Base):
    __tablename__ = 'users'

    id: int = Column(Integer, primary_key=True)
    name: str = Column(String)
    email = Column(String)

    def __init__(self, name):
        self.name = name
        self.email = '[email protected]'


Base.metadata.create_all(engine)

SessionMaker = sessionmaker(bind=engine)
session = SessionMaker()

user1 = User("anurag")
session.add(user1)
session.commit()

query_result = session.query(User).one()  # type: User
print(f'{query_result.id=:}, {query_result.name=:}, {query_result.email=:}')
# query_result.id=1, query_result.name=anurag, [email protected]

query_result_dict = asdict(query_result)
print(query_result_dict)
# {'id': 1, 'name': 'anurag'}

La clave es usar el @dataclassdecorador y anotar cada columna con su tipo (la : strparte de la name: str = Column(String)línea).

También tenga en cuenta que, dado que emailno está anotado, no está incluido en query_result_dict.

Soy un programador de Python recién creado y tuve problemas para llegar a JSON con tablas unidas. Utilizando la información de las respuestas aquí, construí una función para devolver resultados razonables a JSON, donde se incluyen los nombres de las tablas para evitar tener alias o hacer que los campos colisionen.

Simplemente pase el resultado de una consulta de sesión:

test = Session (). query (VMInfo, Customer) .join (Customer) .order_by (VMInfo.vm_name) .limit (50) .offset (10)

json = sqlAl2json (prueba)

def sqlAl2json(self, result):
    arr = []
    for rs in result.all():
        proc = []
        try:
            iterator = iter(rs)
        except TypeError:
            proc.append(rs)
        else:
            for t in rs:
                proc.append(t)

        dict = {}
        for p in proc:
            tname = type(p).__name__
            for d in dir(p):
                if d.startswith('_') | d.startswith('metadata'):
                    pass
                else:
                    key = '%s_%s' %(tname, d)
                    dict[key] = getattr(p, d)
        arr.append(dict)
    return json.dumps(arr)

si la columna de la tabla de modelos no es equie mysql column.

como :

class People:
    id: int = Column(name='id', type_=Integer, primary_key=True)
    createdTime: datetime = Column(name='create_time', type_=TIMESTAMP,
                               nullable=False,
                               server_default=text("CURRENT_TIMESTAMP"),
                               default=func.now())
    modifiedTime: datetime = Column(name='modify_time', type_=TIMESTAMP,
                                server_default=text("CURRENT_TIMESTAMP"),
                                default=func.now())

Necesitará usar:

 from sqlalchemy.orm import class_mapper 
 def asDict(self):
        return {x.key: getattr(self, x.key, None) for x in
            class_mapper(Application).iterate_properties}

si lo usa de esta manera, puede obtener modify_time y create_time, ambos son Ninguno

{'id': 1, 'create_time': None, 'modify_time': None}


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

Porque el nombre de los atributos de clase no es igual al almacén de columnas en mysql

Devuelve el contenido de esto: clase: .KeyedTuplecomo diccionario

In [46]: result = aggregate_events[0]

In [47]: type(result)
Out[47]: sqlalchemy.util._collections.result

In [48]: def to_dict(query_result=None):
    ...:     cover_dict = {key: getattr(query_result, key) for key in query_result.keys()}
    ...:     return cover_dict
    ...: 
    ...:     

In [49]: to_dict(result)
Out[49]: 
{'calculate_avg': None,
 'calculate_max': None,
 'calculate_min': None,
 'calculate_sum': None,
 'dataPointIntID': 6,
 'data_avg': 10.0,
 'data_max': 10.0,
 'data_min': 10.0,
 'data_sum': 60.0,
 'deviceID': u'asas',
 'productID': u'U7qUDa',
 'tenantID': u'CvdQcYzUM'}

Por el bien de todos y de mí, así es como lo uso:

def run_sql(conn_String):
  output_connection = engine.create_engine(conn_string, poolclass=NullPool).connect()
  rows = output_connection.execute('select * from db1.t1').fetchall()  
  return [dict(row) for row in rows]

def to_dict(row):
    return {column.name: getattr(row, row.__mapper__.get_property_by_column(column).key) for column in row.__table__.columns}


for u in session.query(User).all():
    print(to_dict(u))

Esta función puede ayudar. No puedo encontrar una mejor solución para resolver el problema cuando el nombre del atributo es diferente a los nombres de columna.

Lo necesitará en todas partes en su proyecto, apriciado @anurag respondió que funciona bien. hasta este punto lo estaba usando, pero desordenará todo su código y tampoco funcionará con el cambio de entidad.

Más bien intente esto, herede su clase de consulta base en SQLAlchemy

from flask_sqlalchemy import SQLAlchemy, BaseQuery


class Query(BaseQuery):
    def as_dict(self):
        context = self._compile_context()
        context.statement.use_labels = False
        columns = [column.name for column in context.statement.columns]

        return list(map(lambda row: dict(zip(columns, row)), self.all()))


db = SQLAlchemy(query_class=Query)

después de eso, siempre que definas tu objeto, el método "as_dict" estará allí.

En la mayoría de los escenarios, el nombre de la columna es adecuado para ellos. Pero tal vez escriba el código de la siguiente manera:

class UserModel(BaseModel):
    user_id = Column("user_id", INT, primary_key=True)
    email = Column("user_email", STRING)

el column.name "user_email" mientras que el nombre del campo es "email", el column.name no podía funcionar tan bien como antes.

sqlalchemy_base_model.py

También escribo la respuesta aquí