python - serialize - sqlalchemy result as json



Converter o objeto de linha sqlalchemy em python dict (20)

A expressão que você está iterando avalia a lista de objetos do modelo, não as linhas. Então, o seguinte é o uso correto deles:

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

Você realmente precisa convertê-los em ditos? Claro, há muitas 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 módulo sqlalchemy.orm.attributes . Ele tem um conjunto de funções para trabalhar com o estado do objeto, que pode ser útil para você, especialmente instance_dict() .

https://src-bin.com

Existe uma maneira simples de iterar sobre pares de nome e valor de coluna?

Minha versão do sqlalchemy é 0.5.6

Aqui está o código de exemplo onde eu tentei usar o dict (row), mas lança exceção, TypeError: 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)

Executando este 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

Answer #1

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

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

Answer #2

Consulte a resposta da 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 de Alex Brasetvik's Answer, zzzeek o criador de SQLAlchemy afirmou que este é o "Método Correto" para o problema.


Answer #3

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

Answer #4

Eu não consegui uma boa resposta, então eu uso isso:

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 é um liner (python 2.7+)

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

Answer #5

Eu não tenho muita experiência com isso, mas o seguinte parece funcionar para o que estou fazendo:

dict(row)

Isso parece muito simples (comparado com as outras respostas aqui). o que estou perdendo?


Answer #6

Eu tenho uma variação na resposta de Marco Mariani, expressa como decorador. A principal diferença é que ele manipula listas de entidades, assim como ignora com segurança alguns outros tipos de valores de retorno (o que é muito útil ao escrever testes usando 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

Answer #7

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

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 é "email", o column.name não funcionou bem como antes.

sqlalchemy_base_model.py

também eu escrevo a resposta aqui


Answer #8

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)

Answer #9

Para o bem de todos e de mim, aqui está como eu 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]

Answer #10

Retornar o conteúdo deste: class: .KeyedTuple como um dicionário

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

Answer #11

Seguindo a resposta @balki, desde o SQLAlchemy 0.8 você pode usar _asdict (), disponível para objetos KeyedTuple. Isso apresenta uma resposta bastante simples para a pergunta original. Apenas mude em seu exemplo as duas últimas linhas (o loop for) para esta:

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

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

WRT a resposta por @STB: AFAIK, anithong que .all () retorna é uma lista de KeypedTuple. Portanto, o acima funciona ou se você especificar uma coluna ou não, contanto que você esteja lidando com o resultado de .all () aplicado a um objeto Consulta.


Answer #12

Uma solução que também funciona com classes herdadas:

from itertools import chain
from sqlalchemy.ext.declarative import declarative_base
Base = declarative_base()


class Mixin(object):
    def as_dict(self):
        tables = [base.__table__ for base in self.__class__.__bases__ if base not in [Base, Mixin]]
        tables.append(self.__table__)
        return {c.name: getattr(self, c.name) for c in chain.from_iterable([x.columns for x in tables])}

Answer #13

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

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

Answer #14

Você poderia tentar fazer isso 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 dictonary do objeto de consulta.

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


Answer #15

como @balki mencionado:

O método _asdict() 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. Note que este 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'}

Answer #16

se sua coluna de tabela de modelos não for a coluna equie 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, você pode obter modify_time e create_time ambos são Nenhum

{'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 de atributos de classe não é igual ao armazenamento de colunas no mysql


Answer #17

Com este código você também pode adicionar à sua consulta "filter" ou "join" e esse 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]

Answer #18
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.


Answer #19
for row in resultproxy:
    row_as_dict = dict(row)




sqlalchemy