Converter objeto de linha sqlalchemy em dict python

Existe uma maneira simples de iterar sobre o nome da coluna e os pares de valores?

Minha versão do sqlalchemy é 0.5.6

Aqui está o código de exemplo em que tentei usar o dict (linha), mas ele lança a exceção TypeError: o objeto 'User' não é iterável

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)

A execução deste código nas saídas do meu 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

Você pode acessar o interno __dict__de um objeto SQLAlchemy, como o seguinte:

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

Como não consegui uma boa resposta, uso o seguinte:

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

    return d

Edit: se a função acima é muito longa e não é adequada para alguns gostos, aqui está um liner (python 2.7+)

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

Conforme @zzzeek nos comentários:

observe que esta é a resposta correta para versões modernas do SQLAlchemy, supondo que "linha" seja um objeto de linha principal, não uma instância mapeada em ORM.

for row in resultproxy:
    row_as_dict = dict(row)

No SQLAlchemy v0.8 e mais recente, use o sistema de inspeção .

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)

Observe que esse .keyé o nome do atributo, que pode ser diferente do nome da coluna, por exemplo, no seguinte caso:

class_ = Column('class', Text)

Este método também funciona para column_property.

linhas têm uma _asdict()função que fornece um ditado

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 mencionado:

O _asdict()método pode ser usado se você estiver consultando um campo específico porque ele é retornado como um KeyedTuple .

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

Considerando que, se você não especificar uma coluna, poderá usar um dos outros métodos propostos - como o fornecido por @charlax. Observe que esse método é válido apenas 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'}

Pergunta antiga, mas como esse é o primeiro resultado para "ditar sqlalchemy row" no Google, ele merece uma resposta melhor.

O objeto RowProxy que SqlAlchemy retorna possui o método items (): http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items

Ele simplesmente retorna uma lista de tuplas (chave, valor). Portanto, é possível converter uma linha em ditado usando o seguinte:

No Python <= 2.6:

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

Em Python> = 2.7:

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

Assumindo que as seguintes funções serão adicionadas, class Usero seguinte retornará todos os pares de valores-chave de todas as colunas:

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

ao contrário das outras respostas, todos, exceto apenas os atributos do objeto, são retornados, que são Columnatributos no nível de classe do objeto. Portanto, nenhum _sa_instance_stateou qualquer outro atributo SQLalchemy ou você adiciona ao objeto estão incluídos. Referência

Edição: esqueça de dizer que isso também funciona em colunas herdadas.

hybrid_propery extensão

Se você também deseja incluir hybrid_propertyatributos, o seguinte funcionará:

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_

Suponho aqui que você marque as colunas com um começo _para indicar que deseja ocultá-las, porque você acessa o atributo por um hybrid_propertyou simplesmente não deseja mostrá-las. Referência

Tipp all_orm_descriptors também retorna hybrid_method e AssociationProxy, se você também quiser incluí-los.

Comentários a outras respostas

Toda resposta (como 1 , 2 ) que, com base no __dict__atributo, simplesmente retorna todos os atributos do objeto. Isso pode ser muito mais atributos do que você deseja. Como eu triste isso inclui_sa_instance_state ou qualquer outro atributo que você definir sobre esse objeto.

Todas as respostas (como 1 , 2 ) baseadas na dict()função funcionam apenas nos objetos de linha SQLalchemy retornados, session.execute()não nas classes com as quais você define trabalhar, como as class Userda pergunta.

A resposta de solução que se baseia row.__table__.columnsdefinitivamente não funcionará. row.__table__.columnscontém os nomes das colunas do banco de dados SQL. Eles podem ser iguais apenas ao nome dos atributos do objeto python. Se não, você recebe um AttributeError. Para respostas (como 1 , 2 ), com base class_mapper(obj.__class__).mapped_table.cnela, é a mesma.

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)

Após a resposta @balki, desde SQLAlchemy 0.8, você pode usar _asdict (), disponível para objetos KeyedTuple. Isso torna uma resposta bastante direta à pergunta original. Apenas mude no seu exemplo as duas últimas linhas (o loop for) desta:

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

Isso funciona porque no código acima u é um objeto da classe tipo KeyedTuple , pois .all () retorna uma lista de KeyedTuple. Portanto, ele tem o método _asdict () , que retorna u como um dicionário.

Escreva a resposta por @STB: AFAIK, ao lado de .all () retorna uma lista de KeypedTuple. Portanto, o acima funcionará se você especificar uma coluna ou não, desde que esteja lidando com o resultado de .all () aplicado a um objeto de Consulta.

A expressão pela qual você está iterando é avaliada na lista de objetos de modelo , não em linhas. Portanto, o seguinte é o uso correto deles:

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

Você realmente precisa convertê-los em ditados? Claro, existem várias maneiras, mas você não precisa da parte ORM do SQLAlchemy:

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

Atualização : Dê uma olhada no sqlalchemy.orm.attributesmódulo. Ele possui um conjunto de funções para trabalhar com o estado do objeto, que podem ser úteis para você, especialmente instance_dict().

Consulte a resposta de Alex Brasetvik , você pode usar uma linha de código para resolver o problema

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

Sob a seção de comentários da resposta de Alex Brasetvik, zzzeek, ​​o criador do SQLAlchemy, afirmou que este é o "método correto" para o problema.

Você pode tentar fazer dessa maneira.

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

Ele usa um método interno no objeto de consulta que retorna um objeto dictonário do objeto de consulta.

referências: https://docs.sqlalchemy.org/en/latest/orm/query.html

Encontrei este post porque estava procurando uma maneira de converter uma linha SQLAlchemy em um ditado. Estou usando o SqlSoup ... mas a resposta foi criada por mim, então, se isso puder ajudar alguém, aqui estão meus dois centavos:

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

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

Você pode converter o objeto sqlalchemy em dicionário como este e retorná-lo como json / dictionary.

Funções auxiliares:

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) 

Função do driver:

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

Dois caminhos:

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

Os documentos oferecem uma solução muito simples: ResultRow._asdict()

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

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

Aqui está como o Elixir faz isso. O valor dessa solução é que ela permite incluir recursivamente a representação das relações no dicionário.

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

Com este código, você também pode adicionar à sua consulta "filtro" ou "associação" e este trabalho!

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

Isso deve funcionar.

Eu tenho uma variação na resposta de Marco Mariani, expressa como decoradora. A principal diferença é que ele manipulará listas de entidades, além de ignorar com segurança outros tipos de valores de retorno (o que é muito útil ao escrever testes usando zombarias):

@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 a resposta de @Anurag Uniyal, aqui está um método que seguirá recursivamente os relacionamentos:

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}

Sou um programador Python recém-formado e tive problemas para acessar o JSON com tabelas Associadas. Usando informações das respostas aqui, criei uma função para retornar resultados razoáveis ​​ao JSON, onde os nomes das tabelas estão incluídos, evitando a necessidade de um alias ou a colisão de campos.

Basta passar o resultado de uma consulta de sessão:

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

json = sqlAl2json (teste)

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)

se a coluna da tabela de modelos não for igual à coluna mysql.

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

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

se você usar dessa maneira, poderá obter modify_time e create_time, ambos são 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}

Porque o nome dos atributos da classe não é igual ao armazenamento da coluna no mysql

Retorne o conteúdo deste: class: .KeyedTupleas a dictionary

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

Para o bem de todos e de mim, aqui está como eu o 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 função pode ajudar. Não consigo encontrar uma solução melhor para resolver o problema quando o nome do atributo é diferente dos nomes das colunas.

Você precisará dele em qualquer lugar do seu projeto, eu apriciate @anurag respondeu que funciona bem. até o momento, eu estava usando, mas ele bagunçará todo o seu código e também não funcionará com a alteração de entidade.

Em vez disso, tente herdar sua classe de consulta base em 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)

depois disso, onde quer que você defina seu método "as_dict", o objeto estará lá.

Com o python 3.8+, podemos fazer isso com a classe de dados e o asdictmétodo que acompanha:

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

A chave é usar o @dataclassdecorador e anotar cada coluna com seu tipo (a : strparte doname: str = Column(String) linha).

Observe também que, como o emailnão é anotado, ele não é incluído no query_result_dict.

Na maioria dos cenários, o nome da coluna é adequado para eles. Mas talvez você escreva o código da seguinte maneira:

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

o column.name "user_email" enquanto o nome do campo for "email", o column.name não funcionou bem como antes.

sqlalchemy_base_model.py

Também escrevo a resposta aqui