Konvertieren Sie das sqlalchemy-Zeilenobjekt in Python dict

Gibt es eine einfache Möglichkeit, über Spaltennamen- und Wertepaare zu iterieren?

Meine Version von sqlalchemy ist 0.5.6

Hier ist der Beispielcode, in dem ich versucht habe, dict (row) zu verwenden, aber er löst die Ausnahme TypeError aus: Das Objekt 'User' ist nicht iterierbar

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)

Ausführen dieses Codes auf meinen Systemausgaben:

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

Sie können auf das Interne __dict__eines SQLAlchemy-Objekts zugreifen , wie folgt:

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

Ich konnte keine gute Antwort bekommen und benutze diese:

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

    return d

Bearbeiten: Wenn die obige Funktion zu lang ist und für einige Geschmäcker nicht geeignet ist, ist hier ein Einzeiler (Python 2.7+)

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

Laut @zzzeek in Kommentaren:

Beachten Sie, dass dies die richtige Antwort für moderne Versionen von SQLAlchemy ist, vorausgesetzt, "row" ist ein Kernzeilenobjekt und keine ORM-zugeordnete Instanz.

for row in resultproxy:
    row_as_dict = dict(row)

Verwenden Sie in SQLAlchemy v0.8 und höher das Inspektionssystem .

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)

Beachten Sie, dass dies .keyder Attributname ist, der sich vom Spaltennamen unterscheiden kann, z. B. im folgenden Fall:

class_ = Column('class', Text)

Diese Methode funktioniert auch für column_property.

Zeilen haben eine _asdict()Funktion, die ein Diktat gibt

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

wie @balki erwähnt:

Die _asdict()Methode kann verwendet werden, wenn Sie ein bestimmtes Feld abfragen, da es als KeyedTuple zurückgegeben wird .

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

Wenn Sie keine Spalte angeben, können Sie eine der anderen vorgeschlagenen Methoden verwenden - beispielsweise die von @charlax bereitgestellte. Beachten Sie, dass diese Methode nur für 2.7+ gültig ist.

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

Alte Frage, aber da dies das erste Ergebnis für "sqlalchemy row to dict" in Google ist, verdient es eine bessere Antwort.

Das von SqlAlchemy zurückgegebene RowProxy-Objekt verfügt über die Methode items (): http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items

Es wird einfach eine Liste von (Schlüssel-, Wert-) Tupeln zurückgegeben. So kann man eine Zeile wie folgt in ein Diktat umwandeln:

In Python <= 2.6:

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

In Python> = 2.7:

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

Angenommen, die folgenden Funktionen werden zu den folgenden hinzugefügt, class Userwerden alle Schlüssel-Wert-Paare aller Spalten zurückgegeben:

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

Im Gegensatz zu den anderen Antworten werden alle Attribute außer dem Objekt zurückgegeben, bei denen es sich um ColumnAttribute auf Klassenebene des Objekts handelt. Daher sind keine _sa_instance_stateoder andere Attribute SQLalchemy oder Sie, die Sie dem Objekt hinzufügen, enthalten. Referenz

EDIT: Vergessen Sie zu sagen, dass dies auch für geerbte Spalten funktioniert.

hybrid_propery Verlängerung

Wenn Sie auch hybrid_propertyAttribute einschließen möchten, funktioniert Folgendes:

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_

Ich gehe hier davon aus, dass Sie Spalten mit einem Anfang _markieren, um anzuzeigen, dass Sie sie ausblenden möchten, entweder weil Sie über ein auf das Attribut zugreifen hybrid_propertyoder weil Sie sie einfach nicht anzeigen möchten . Referenz

Tipp gibt all_orm_descriptors auch hybrid_method und AssociationProxy zurück, wenn Sie sie ebenfalls einschließen möchten.

Anmerkungen zu anderen Antworten

Jede Antwort (wie 1 , 2 ), die auf dem __dict__Attribut basiert , gibt einfach alle Attribute des Objekts zurück. Dies können viel mehr Attribute sein, als Sie möchten. Wie ich traurig finde, enthält dies _sa_instance_stateoder jedes andere Attribut, das Sie für dieses Objekt definieren.

Jede Antwort (wie 1 , 2 ), die auf der dict()Funktion basiert, funktioniert nur mit SQLalchemy-Zeilenobjekten, die von session.execute()nicht den Klassen zurückgegeben werden, mit denen Sie arbeiten möchten , wie z. B. class Useraus der Frage.

Die Antwort auf die Lösung, auf der basiert, row.__table__.columnswird definitiv nicht funktionieren. row.__table__.columnsenthält die Spaltennamen der SQL-Datenbank. Diese können nur dem Attributnamen des Python-Objekts entsprechen. Wenn nicht, bekommst du eine AttributeError. Für Antworten (wie 1 , 2 ), die darauf basieren class_mapper(obj.__class__).mapped_table.c, ist es dasselbe.

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)

Nach der Antwort von @balki können Sie seit SQLAlchemy 0.8 _asdict () verwenden, das für KeyedTuple-Objekte verfügbar ist. Dies ergibt eine ziemlich einfache Antwort auf die ursprüngliche Frage. Ändern Sie in Ihrem Beispiel einfach die letzten beiden Zeilen (die for-Schleife) für diese:

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

Dies funktioniert, weil im obigen Code u ein Objekt der Typklasse KeyedTuple ist , da .all () eine Liste von KeyedTuple zurückgibt. Daher hat es die Methode _asdict () , die u schön als Wörterbuch zurückgibt.

WRT die Antwort von @STB: AFAIK, obwohl .all () eine Liste von KeypedTuple zurückgibt. Daher funktioniert das Obige entweder, wenn Sie eine Spalte angeben oder nicht, solange Sie sich mit dem Ergebnis von .all () befassen, das auf ein Abfrageobjekt angewendet wird.

Der Ausdruck, den Sie durchlaufen, wird zu einer Liste von Modellobjekten und nicht zu Zeilen ausgewertet . Das Folgende ist also die richtige Verwendung von ihnen:

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

Müssen Sie sie wirklich in Diktate umwandeln? Sicher, es gibt viele Möglichkeiten, aber dann brauchen Sie keinen ORM-Teil von SQLAlchemy:

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

Update : Schauen Sie sich das sqlalchemy.orm.attributesModul an. Es verfügt über eine Reihe von Funktionen zum Arbeiten mit dem Objektstatus, die insbesondere für Sie nützlich sein können instance_dict().

In der Antwort von Alex Brasetvik können Sie eine Codezeile verwenden, um das Problem zu lösen

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

Im Kommentarbereich von Alex Brasetviks Antwort erklärte zzzeek, ​​der Schöpfer von SQLAlchemy, dies sei die "richtige Methode" für das Problem.

Sie könnten versuchen, es auf diese Weise zu tun.

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

Es verwendet eine im Abfrageobjekt integrierte Methode, die ein diktonisches Objekt des Abfrageobjekts zurückgibt.

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

Ich habe diesen Beitrag gefunden, weil ich nach einer Möglichkeit gesucht habe, eine SQLAlchemy-Zeile in ein Diktat umzuwandeln. Ich benutze SqlSoup ... aber die Antwort wurde von mir selbst erstellt. Wenn es jemandem helfen könnte, sind hier meine zwei Cent:

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

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

Sie können das sqlalchemy-Objekt wie folgt in ein Wörterbuch konvertieren und es als json / dictionary zurückgeben.

Hilfsfunktionen:

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) 

Treiberfunktion:

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

Zwei Wege:

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

Die Dokumente bieten eine sehr einfache Lösung: ResultRow._asdict()

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

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

So macht es Elixir. Der Wert dieser Lösung besteht darin, dass sie die rekursive Einbeziehung der Wörterbuchdarstellung von Beziehungen ermöglicht.

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

Mit diesem Code können Sie Ihrer Abfrage auch "Filter" oder "Join" hinzufügen und diese Arbeit!

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

Das sollte funktionieren.

Ich habe eine Variation von Marco Marianis Antwort, ausgedrückt als Dekorateur. Der Hauptunterschied besteht darin, dass Listen von Entitäten verarbeitet und einige andere Arten von Rückgabewerten ignoriert werden (was beim Schreiben von Tests mit Mocks sehr nützlich ist):

@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

Um die Antwort von @Anurag Uniyal zu vervollständigen, folgt eine Methode, die rekursiv Beziehungen folgt:

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}

Ich bin ein neu geprägter Python-Programmierer und hatte Probleme, mit verbundenen Tabellen zu JSON zu gelangen. Unter Verwendung der Informationen aus den Antworten hier habe ich eine Funktion erstellt, um angemessene Ergebnisse an JSON zurückzugeben, in der die Tabellennamen enthalten sind, ohne dass ein Alias ​​erforderlich ist oder Felder kollidieren.

Übergeben Sie einfach das Ergebnis einer Sitzungsabfrage:

test = Session (). query (VMInfo, Kunde) .join (Kunde) .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)

Wenn Ihre Modelltabellenspalte nicht der MySQL-Spalte entspricht.

sowie :

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

Benötigt werden:

 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}

Wenn Sie diese Methode verwenden, können Sie "modify_time" und "create_time" abrufen. Beide sind "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}

Weil der Name der Klassenattribute nicht mit dem Spaltenspeicher in MySQL übereinstimmt

Geben Sie den Inhalt dieser: class: .KeyedTupleals Wörterbuch zurück

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

Für alle und mich selbst benutze ich Folgendes:

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

Diese Funktion könnte helfen. Ich kann keine bessere Lösung finden, um das Problem zu lösen, wenn sich der Attributname von den Spaltennamen unterscheidet.

Sie werden es überall in Ihrem Projekt brauchen. Ich schätze @anurag antwortete, dass es gut funktioniert. Bis zu diesem Punkt habe ich es verwendet, aber es wird Ihren gesamten Code durcheinander bringen und auch nicht mit Entitätsänderungen funktionieren.

Versuchen Sie dies lieber und erben Sie Ihre Basisabfrageklasse in 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)

Danach ist überall dort, wo Sie Ihr Objekt definieren, die Methode "as_dict" vorhanden.

Mit Python 3.8+ können wir dies mit der Datenklasse und der damit verbundenen asdictMethode tun :

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

Der Schlüssel besteht darin, den @dataclassDekorator zu verwenden und jede Spalte mit ihrem Typ (dem : strTeil der name: str = Column(String)Zeile) zu versehen.

Beachten Sie auch, dass das emailnicht mit Anmerkungen versehen ist und nicht in enthalten ist query_result_dict.

In den meisten Szenarien ist der Spaltenname für sie geeignet. Aber vielleicht schreiben Sie den Code wie folgt:

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

Der Spaltenname "user_email", während der Feldname "email" lautet, konnte der Spaltenname nicht wie zuvor funktionieren.

sqlalchemy_base_model.py

auch ich schreibe die antwort hier