Convertir un objet ligne sqlalchemy en dict python

Existe-t-il un moyen simple d'itérer les paires nom de colonne et valeur?

Ma version de sqlalchemy est 0.5.6

Voici l'exemple de code où j'ai essayé d'utiliser dict (row), mais il lève une exception, TypeError: l'objet 'User' n'est pas itérable

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)

Exécuter ce code sur mes sorties système:

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

Vous pouvez accéder à l'intérieur __dict__d'un objet SQLAlchemy, comme suit:

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

Je n'ai pas pu obtenir une bonne réponse, donc j'utilise ceci:

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

    return d

Edit: si la fonction ci-dessus est trop longue et ne convient pas à certains goûts voici un one liner (python 2.7+)

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

Selon @zzzeek dans les commentaires:

notez que c'est la bonne réponse pour les versions modernes de SQLAlchemy, en supposant que "row" est un objet de ligne principal, pas une instance mappée ORM.

for row in resultproxy:
    row_as_dict = dict(row)

Dans SQLAlchemy v0.8 et plus récent, utilisez le système d'inspection .

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)

Notez qu'il .keys'agit du nom de l'attribut, qui peut être différent du nom de la colonne, par exemple dans le cas suivant:

class_ = Column('class', Text)

Cette méthode fonctionne également pour column_property.

les lignes ont une _asdict()fonction qui donne 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'}

comme @balki l'a mentionné:

La _asdict()méthode peut être utilisée si vous interrogez un champ spécifique car il est renvoyé en tant que KeyedTuple .

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

Alors que si vous ne spécifiez pas de colonne, vous pouvez utiliser l'une des autres méthodes proposées - comme celle fournie par @charlax. Notez que cette méthode n'est valide que pour la version 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'}

Vieille question, mais comme c'est le premier résultat pour "sqlalchemy row to dict" dans Google, il mérite une meilleure réponse.

L'objet RowProxy que SqlAlchemy renvoie a la méthode items (): http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items

Il retourne simplement une liste de tuples (clé, valeur). On peut donc convertir une ligne en dict en utilisant ce qui suit:

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]

En supposant que les fonctions suivantes seront ajoutées à la class Usersuivante, toutes les paires valeur / clé de toutes les colonnes seront renvoyées:

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

contrairement aux autres réponses, tous les attributs de l'objet sont renvoyés, mais uniquement ceux qui sont des Columnattributs au niveau de la classe de l'objet. Par conséquent, aucun _sa_instance_stateou tout autre attribut SQLalchemy ou que vous ajoutez à l'objet n'est inclus. Référence

EDIT: Oubliez de dire que cela fonctionne également sur les colonnes héritées.

hybrid_propery extention

Si vous souhaitez également inclure des hybrid_propertyattributs, les éléments suivants fonctionneront:

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_

Je suppose ici que vous marquez les colonnes avec un début _pour indiquer que vous souhaitez les masquer, soit parce que vous accédez à l'attribut par hybrid_propertyou vous ne voulez tout simplement pas les afficher. Référence

Tipp all_orm_descriptors renvoie également hybrid_method et AssociationProxy si vous souhaitez également les inclure.

Remarques sur d'autres réponses

Chaque réponse (comme 1 , 2 ) qui, sur la base de l' __dict__attribut, renvoie simplement tous les attributs de l'objet. Cela pourrait être beaucoup plus d'attributs que vous le souhaitez. Comme je le regrette, cela inclut _sa_instance_stateou tout autre attribut que vous définissez sur cet objet.

Chaque réponse (comme 1 , 2 ) basée sur la dict()fonction ne fonctionne que sur les objets de ligne SQLalchemy renvoyés par session.execute()les classes que vous définissez pour travailler, comme class Userdans la question.

La réponse de résolution basée sur row.__table__.columnsne fonctionnera certainement pas . row.__table__.columnscontient les noms de colonne de la base de données SQL. Ceux- ci ne peuvent être égaux qu'au nom d'attributs de l'objet python. Sinon, vous obtenez un AttributeError. Pour les réponses (comme 1 , 2 ) basées sur class_mapper(obj.__class__).mapped_table.cc'est la même chose.

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)

Suite à la réponse de @balki, depuis SQLAlchemy 0.8, vous pouvez utiliser _asdict (), disponible pour les objets KeyedTuple. Cela rend une réponse assez simple à la question d'origine. Modifiez simplement dans votre exemple les deux dernières lignes (la boucle for) pour celle-ci:

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

Cela fonctionne parce que dans le code ci-dessus, u est un objet de type classe KeyedTuple , car .all () renvoie une liste de KeyedTuple. Par conséquent, il a la méthode _asdict () , qui retourne joliment u en tant que dictionnaire.

WRT la réponse de @STB: AFAIK, une fois que .all () retourne est une liste de KeypedTuple. Par conséquent, ce qui précède fonctionne si vous spécifiez une colonne ou non, tant que vous traitez le résultat de .all () appliqué à un objet Query.

L'expression que vous parcourez évalue la liste des objets du modèle , pas les lignes. Ainsi, leur utilisation est correcte:

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

Avez-vous vraiment besoin de les convertir en dict? Bien sûr, il existe de nombreuses façons, mais vous n'avez pas besoin de la partie ORM de SQLAlchemy:

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

Mise à jour : jetez un œil au sqlalchemy.orm.attributesmodule. Il dispose d'un ensemble de fonctions pour travailler avec l'état de l'objet, qui pourraient vous être utiles, en particulier instance_dict().

Référez-vous à la réponse d'Alex Brasetvik , vous pouvez utiliser une ligne de code pour résoudre le problème

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

Dans la section des commentaires de la réponse d'Alex Brasetvik, zzzeek, ​​le créateur de SQLAlchemy, a déclaré que c'était la "bonne méthode" pour le problème.

Vous pouvez essayer de le faire de cette façon.

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

Il utilise une méthode intégrée dans l'objet de requête qui renvoie un objet dictonary de l'objet de requête.

références: https://docs.sqlalchemy.org/en/latest/orm/query.html

J'ai trouvé ce message parce que je cherchais un moyen de convertir une ligne SQLAlchemy en dict. J'utilise SqlSoup ... mais la réponse a été construite par moi-même, donc, si cela peut aider quelqu'un, voici mes deux cents:

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

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

Vous pouvez convertir un objet sqlalchemy en dictionnaire comme celui-ci et le renvoyer en tant que json / dictionary.

Fonctions d'assistance:

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) 

Fonction pilote:

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

Deux manières:

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

Les documents offrent une solution très 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)

Voici comment Elixir le fait. L'intérêt de cette solution est qu'elle permet d'inclure récursivement la représentation dictionnaire des relations.

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

Avec ce code, vous pouvez également ajouter à votre requête "filtre" ou "rejoindre" et ce travail!

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("_")])

Cela devrait fonctionner.

J'ai une variation sur la réponse de Marco Mariani, exprimée en tant que décorateur. La principale différence est qu'il gère les listes d'entités, tout en ignorant en toute sécurité certains autres types de valeurs de retour (ce qui est très utile lors de l'écriture de tests à l'aide de mocks):

@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

Pour compléter la réponse de @Anurag Uniyal, voici une méthode qui suivra récursivement les relations:

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}

Avec python 3.8+, nous pouvons le faire avec la classe de données et la asdictméthode qui l'accompagne:

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

L'essentiel est d'utiliser le @dataclassdécorateur et d'annoter chaque colonne avec son type (la : strpartie de la name: str = Column(String)ligne).

Notez également que puisque le emailn'est pas annoté, il n'est pas inclus dans query_result_dict.

Je suis un programmeur Python nouvellement créé et j'ai rencontré des problèmes pour accéder à JSON avec les tables jointes. En utilisant les informations des réponses ici, j'ai construit une fonction pour renvoyer des résultats raisonnables à JSON où les noms de table sont inclus en évitant d'avoir à créer un alias ou à faire entrer les champs en collision.

Passez simplement le résultat d'une requête de session:

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

json = sqlAl2json (test)

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 colonne de votre table de modèles n'est pas la colonne equie mysql.

tel que :

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

Besoin d'utiliser:

 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 vous utilisez cette méthode, vous pouvez obtenir modify_time et create_time les deux sont None

{'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}

Parce que le nom des attributs de classe n'est pas égal au magasin de colonnes dans mysql

Renvoie le contenu de ceci: classe: .KeyedTuplesous forme de dictionnaire

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

Pour le bien de tous et de moi-même, voici comment je l'utilise:

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

Cette fonction pourrait aider. Je ne peux pas trouver de meilleure solution pour résoudre le problème lorsque le nom d'attribut est différent des noms de colonne.

Vous en aurez besoin partout dans votre projet, je apriciate @anurag a répondu que cela fonctionne bien. jusqu'à présent, je l'utilisais, mais il gâchera tout votre code et ne fonctionnera pas non plus avec le changement d'entité.

Essayez plutôt ceci, héritez de votre classe de requête de base dans 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)

après cela, partout où vous définirez votre méthode d'objet "as_dict" sera là.

Dans la plupart des scénarios, le nom de la colonne leur convient. Mais peut-être que vous écrivez le code comme suit:

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

le column.name "user_email" tandis que le nom du champ est "email", le column.name ne pouvait pas bien fonctionner comme avant.

sqlalchemy_base_model.py

aussi j'écris la réponse ici