Konwertuj obiekt wiersza sqlalchemy na Python Dict

Czy istnieje prosty sposób na iterację par nazw kolumn i wartości?

Moja wersja sqlalchemy to 0.5.6

Oto przykładowy kod, w którym próbowałem użyć dict (wiersz), ale zgłasza wyjątek, TypeError: Obiekt „User” nie jest iterowalny

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)

Uruchomienie tego kodu na wyjściach systemu:

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

Możesz uzyskać dostęp do wnętrza __dict__obiektu SQLAlchemy, na przykład:

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

Nie mogłem uzyskać dobrej odpowiedzi, więc używam tego:

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

    return d

Edycja: jeśli powyższa funkcja jest zbyt długa i nie nadaje się do niektórych smaków, to jest to jedna linijka (python 2.7+)

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

Jak na @zzzeek w komentarzach:

zauważ, że jest to poprawna odpowiedź dla współczesnych wersji SQLAlchemy, zakładając, że „row” jest głównym obiektem wiersza, a nie instancją odwzorowaną na ORM.

for row in resultproxy:
    row_as_dict = dict(row)

W SQLAlchemy v0.8 i nowszych użyj systemu inspekcji .

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)

Zauważ, że .keyjest to nazwa atrybutu, która może różnić się od nazwy kolumny, np. W następującym przypadku:

class_ = Column('class', Text)

Ta metoda działa również w przypadku column_property.

wiersze mają _asdict()funkcję, która daje dyktando

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

jak wspomniał @balki:

_asdict()Metoda może być stosowana, jeśli jesteś zapytań pole specyficzny, ponieważ jest zwracany jako KeyedTuple .

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

Natomiast jeśli nie określisz kolumny, możesz użyć jednej z innych proponowanych metod - takich jak ta dostarczona przez @charlax. Pamiętaj, że ta metoda jest ważna tylko dla wersji 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'}

Stare pytanie, ale ponieważ jest to pierwszy wynik dla „wiersza sqlalchemy do dyktowania” w Google, zasługuje na lepszą odpowiedź.

Obiekt RowProxy, który zwraca SqlAlchemy, ma metodę items (): http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items

Zwraca po prostu listę krotek (klucz, wartość). Aby przekonwertować wiersz na dyktando, można wykonać następujące czynności:

W Pythonie <= 2.6:

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

W Pythonie> = 2.7:

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

Zakładając, że następujące funkcje zostaną dodane do class Usernastępujących, zwrócą wszystkie pary klucz-wartość wszystkich kolumn:

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

w przeciwieństwie do innych odpowiedzi zwracane są tylko te atrybuty obiektu, które są Columnatrybutami na poziomie klasy obiektu. Dlatego nie są uwzględniane _sa_instance_stateżadne inne atrybuty SQLalchemy ani dodane do obiektu. Odniesienie

EDYCJA: Zapomnij powiedzieć, że działa to również na odziedziczone kolumny.

hybrid_propery rozszerzenie

Jeśli chcesz również dołączyć hybrid_propertyatrybuty, następujące działania będą działać:

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_

Zakładam tutaj, że zaznaczasz Kolumny początkiem, _aby wskazać, że chcesz je ukryć, albo dlatego, że uzyskujesz dostęp do atrybutu przez, hybrid_propertyalbo po prostu nie chcesz ich pokazywać. Odniesienie

Tipp all_orm_descriptors zwraca także hybrid_method i AssociationProxy jeśli chcesz je również dołączyć.

Uwagi do innych odpowiedzi

Każda odpowiedź (jak 1 , 2 ), która na podstawie __dict__atrybutu zwraca po prostu wszystkie atrybuty obiektu. Może to być znacznie więcej atrybutów, niż chcesz. To smutne, że to obejmuje_sa_instance_state lub dowolny inny atrybut zdefiniowany w tym obiekcie.

Każda odpowiedź (np. 1 , 2 ), która jest oparta na dict()funkcji, działa tylko na obiektach wierszy SQLalchemy zwracanych przez session.execute()nie na klasach, z którymi zdefiniujesz pracę, takich jakclass User z pytania.

Odpowiedź rozwiązywanie który oparty jest na row.__table__.columnspewno nie praca. row.__table__.columnszawiera nazwy kolumn bazy danych SQL. Mogą być one równe tylko nazwie atrybutu obiektu python. Jeśli nie, dostaniesz AttributeError. Odpowiedzi (jak 1 , 2 ) oparte na class_mapper(obj.__class__).mapped_table.cnim są takie same.

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)

Po odpowiedzi @balki, od SQLAlchemy 0.8 możesz użyć _asdict (), dostępnej dla obiektów KeyedTuple. To daje dość prostą odpowiedź na pierwotne pytanie. Po prostu zmień w swoim przykładzie dwie ostatnie linie (pętlę for) dla tego:

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

Działa to, ponieważ w powyższym kodzie u jest obiektem klasy typu KeyedTuple , ponieważ .all () zwraca listę KeyedTuple. Dlatego ma metodę _asdict () , która ładnie zwraca u jako słownik.

WRT odpowiedź @STB: AFAIK, po czym zwraca .all () jest listą KeypedTuple. Dlatego powyższe działa albo, jeśli podasz kolumnę, albo nie, o ile masz do czynienia z wynikiem .all () zastosowanym do obiektu zapytania.

Wyrażenie, przez które się iteruje, przekształca się w listę obiektów modelu , a nie wiersze. Więc następujące jest ich prawidłowe użycie:

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

Czy naprawdę musisz przekonwertować je na dyktanda? Jasne, istnieje wiele sposobów, ale wtedy nie potrzebujesz części ORM SQLAlchemy:

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

Aktualizacja : spójrz na sqlalchemy.orm.attributesmoduł. Posiada zestaw funkcji do pracy ze stanem obiektu, który może być szczególnie przydatny instance_dict().

Zapoznaj się z odpowiedzią Alexa Brasetvika , aby rozwiązać problem, możesz użyć jednego wiersza kodu

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

W sekcji komentarza w odpowiedzi Alexa Brasetvika zzzeek, ​​twórca SQLAlchemy, stwierdził, że jest to „poprawna metoda” problemu.

Możesz spróbować to zrobić w ten sposób.

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

Używa wbudowanej metody w obiekcie zapytania, która zwraca obiekt dictonary obiektu zapytania.

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

Znalazłem ten post, ponieważ szukałem sposobu na konwersję wiersza SQLAlchemy na dyktando. Używam SqlSoup ... ale odpowiedź została zbudowana przeze mnie, więc jeśli to mogłoby pomóc komuś, oto moje dwa centy:

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

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

Możesz przekonwertować obiekt sqlalchemy na słownik w ten sposób i zwrócić go jako json / dictionary.

Funkcje pomocnicze:

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) 

Funkcja kierowcy:

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

Dwie drogi:

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

Dokumenty oferują bardzo proste rozwiązanie: ResultRow._asdict()

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

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

Oto, jak robi to Elixir. Wartość tego rozwiązania polega na tym, że pozwala on rekurencyjnie uwzględniać słownikową reprezentację relacji.

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

Za pomocą tego kodu możesz również dodać do zapytania „filtr” lub „dołącz” i to działa!

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

To powinno działać.

Mam odmianę odpowiedzi Marco Mariani, wyrażoną jako dekorator. Główną różnicą jest to, że obsługuje listy jednostek, a także bezpiecznie ignoruje niektóre inne typy zwracanych wartości (co jest bardzo przydatne podczas pisania testów przy użyciu próbnych):

@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

Aby wypełnić odpowiedź @Anurag Uniyal, oto metoda, która będzie rekurencyjnie śledzić relacje:

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}

W Pythonie 3.8+ możemy to zrobić za pomocą klasy danych i asdictdołączonej do niej metody:

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 = 'hello@example.com'


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, query_result.email=hello@example.com

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

Kluczem jest użycie @dataclassdekoratora i opatrzenie adnotacjami każdej kolumny swoim typem ( : strczęśćname: str = Column(String) linii).

Pamiętaj też, że ponieważ emailnie jest to adnotacja, nie jest uwzględnione w query_result_dict.

Jestem świeżo upieczonym programistą Python i miałem problemy z dostępem do JSON z dołączonymi tabelami. Korzystając z informacji zawartych w odpowiedziach tutaj, zbudowałem funkcję zwracającą rozsądne wyniki do JSON-a, gdzie zawarte są nazwy tabel, unikając konieczności aliasu lub kolizji pól.

Po prostu przekaż wynik zapytania o sesję:

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)

jeśli kolumna tabeli modeli nie jest równa kolumnie mysql.

Jak na przykład :

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

Potrzebuję użyć:

 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}

jeśli użyjesz tej metody, możesz uzyskać czas modyfikacji i czas utworzenia, oba są Brak

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

Ponieważ nazwa atrybutów klasy nie jest równa nazwie magazynu kolumn w mysql

Zwróć zawartość tego: class: .KeyedTuplejako słownika

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

Ze względu na wszystkich i siebie, oto jak go używam:

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

Ta funkcja może pomóc. Nie mogę znaleźć lepszego rozwiązania problemu, gdy nazwa atrybutu jest inna niż nazwa kolumny.

Będziesz go potrzebować wszędzie w swoim projekcie, apriciate @anurag odpowiedział, że działa dobrze. do tego momentu używałem go, ale to zepsuje cały twój kod, a także nie będzie działać ze zmianą encji.

Zamiast tego spróbuj odziedziczyć podstawową klasę zapytania w 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)

po tym, gdziekolwiek zdefiniujesz swój obiekt, pojawi się metoda „as_dict”.

W większości scenariuszy nazwa kolumny jest dla nich odpowiednia. Ale może piszesz kod w następujący sposób:

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

nazwa.kolumny „email_użytkownika”, podczas gdy nazwa pola to „e-mail”, nazwa.kolumny nie mogła działać tak dobrze jak wcześniej.

sqlalchemy_base_model.py

też piszę odpowiedź tutaj