Преобразовать объект строки sqlalchemy в dict dict
Есть ли простой способ перебирать пары имен и значений столбцов?
Моя версия sqlalchemy 0.5.6
Вот пример кода, где я пытался использовать dict(строку), но он выдает исключение, TypeError: объект "Пользователь" не повторяется
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)
Выполнение этого кода на выходах моей системы:
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
45 ответов
Я не мог получить хороший ответ, поэтому я использую это:
def row2dict(row):
d = {}
for column in row.__table__.columns:
d[column.name] = str(getattr(row, column.name))
return d
Редактировать: если указанная выше функция слишком длинная и не подходит для некоторых вкусов, здесь указывается одна строка (python 2.7+)
row2dict = lambda r: {c.name: str(getattr(r, c.name)) for c in r.__table__.columns}
Обновление 2018-03-25:
Вы больше не можете использовать u.__dict__, поскольку он интерпретирует его как ссылку на столбец / поле. Вместо этого используйте dict(u),
Оригинальный ответ:
Вы можете получить доступ к внутреннему __dict__ объекта SQLAlchemy, например:
for u in session.query(User).all():
print u.__dict__
Это правильный ответ для современной SQLAlchemy (v0.8 - v1.2+).
Используйте систему контроля.
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)
Обратите внимание, что .key это имя атрибута, которое может отличаться от имени столбца, например, в следующем случае:
class_ = Column('class', Text)
Этот метод также работает для column_property,
Строки имеют _asdict() функция, которая дает диктат
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'}
Предполагая, что следующие функции будут добавлены к class User следующее вернет все пары ключ-значение всех столбцов:
def columns_to_dict(self):
dict_ = {}
for key in self.__mapper__.c.keys():
dict_[key] = getattr(self, key)
return dict_
в отличие от других ответов возвращаются все, кроме атрибутов объекта, которые Columnатрибуты на уровне класса объекта. Поэтому нет_sa_instance_stateили любой другой атрибутSQLalchemy или вы добавляете к объекту включены. Ссылка
РЕДАКТИРОВАТЬ: забудьте, что это также работает на унаследованных столбцах.
hybrid_properyпродление
Если вы также хотите включитьhybrid_propertyАтрибуты будут работать следующие:
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_
Я предполагаю, что вы помечаете столбцы началом_чтобы указать, что вы хотите скрыть их, потому что вы получаете доступ к атрибуту с помощьюhybrid_propertyили вы просто не хотите их показывать. Ссылка
Типпall_orm_descriptorsтакже возвращает hybrid_method и AssociationProxy, если вы также хотите включить их.
Замечания к другим ответам
Каждый ответ (например, 1, 2), основанный на __dict__Атрибут просто возвращает все атрибуты объекта. Это может быть гораздо больше атрибутов, чем вы хотите. Как мне грустно это включает_sa_instance_state или любой другой атрибут, который вы определяете для этого объекта.
Каждый ответ (например, 1, 2), который основан на dict()функция работает только собъектами строк SQLalchemy, возвращаемыми session.execute() не на классах, которые вы определили для работы, как class User из вопроса.
Решающий ответ, основанный на row.__table__.columns точно не сработает. row.__table__.columns содержит имена столбцов базы данных SQL. Они могут быть равны только имени атрибута объекта python. Если нет, вы получите AttributeError, Для ответов (например, 1, 2) на основе class_mapper(obj.__class__).mapped_table.c это то же самое.
Как упомянул @balki:
_asdict() Метод можно использовать, если вы запрашиваете определенное поле, потому что оно возвращается как KeyedTuple.
In [1]: foo = db.session.query(Topic.name).first()
In [2]: foo._asdict()
Out[2]: {'name': u'blah'}
Принимая во внимание, что если вы не укажете столбец, вы можете использовать один из предложенных методов, например, предоставленный @charlax. Обратите внимание, что этот метод действителен только для версии 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'}
Старый вопрос, но так как это первый результат для "sqlalchemy row to dict" в Google, он заслуживает лучшего ответа.
Объект RowProxy, который возвращает SqlAlchemy, имеет метод items(): http://docs.sqlalchemy.org/en/latest/core/connections.html
Он просто возвращает список (ключ, значение) кортежей. Таким образом, можно преобразовать строку в dict, используя следующее:
В Python <= 2.6:
rows = conn.execute(query)
list_of_dicts = [dict((key, value) for key, value in row.items()) for row in rows]
В Python >= 2.7:
rows = conn.execute(query)
list_of_dicts = [{key: value for (key, value) in row.items()} for row in rows]
с sqlalchemy 1.4
session.execute(select(User.id, User.username)).mappings().all()
>> [{'id': 1, 'username': 'Bob'}, {'id': 2, 'username': 'Alice'}]
Документы предлагают очень простое решение: ResultRow._asdict()
def to_array(rows):
return [r._asdict() for r in rows]
def query():
data = session.query(Table).all()
return to_array(data)
После ответа @balki, начиная с SQLAlchemy 0.8, вы можете использовать _asdict(), доступный для объектов KeyedTuple. Это дает довольно простой ответ на оригинальный вопрос. Просто измените в вашем примере две последние строки (цикл for) на эту:
for u in session.query(User).all():
print u._asdict()
Это работает, потому что в приведенном выше коде u является объектом класса типа KeyedTuple, поскольку.all() возвращает список KeyedTuple. Поэтому у него есть метод _asdict (), который приятно возвращает вас в качестве словаря.
WRT ответ @STB: AFAIK, и что.all() возвращает список KeypedTuple. Следовательно, вышеприведенное работает либо в том случае, если вы укажете столбец, либо нет, если вы имеете дело с результатом.all() применительно к объекту Query.
Обратитесь к ответу Алекса Брасетвика, вы можете использовать одну строку кода для решения проблемы
row_as_dict = [dict(row) for row in resultproxy]
В разделе комментариев Ответа Алексея Брасетвика, zzzeek, создатель SQLAlchemy, заявил, что это "правильный метод" для решения проблемы.
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)
С python 3.8+ мы можем сделать это с помощью класса данных, а asdict метод, который идет с ним:
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'}
Ключ в том, чтобы использовать @dataclass декоратор и аннотируйте каждый столбец его типом (: str часть name: str = Column(String) линия).
Также обратите внимание, что поскольку email не аннотируется, не входит в query_result_dict.
Я нашел этот пост, потому что искал способ конвертировать строку SQLAlchemy в dict. Я использую SqlSoup... но ответ был создан мной, так что, если это могло бы помочь кому-то, вот мои два цента:
a = db.execute('select * from acquisizioni_motes')
b = a.fetchall()
c = b[0]
# and now, finally...
dict(zip(c.keys(), c.values()))
Выражение, через которое вы перебираете, оценивает список объектов модели, а не строк. Таким образом, следующее правильное использование их:
for u in session.query(User).all():
print u.id, u.name
Вам действительно нужно конвертировать их в диктовку? Конечно, есть много способов, но тогда вам не нужна ORM часть SQLAlchemy:
result = session.execute(User.__table__.select())
for row in result:
print dict(row)
Обновление: посмотрите на sqlalchemy.orm.attributes модуль. Он имеет набор функций для работы с состоянием объекта, которые могут быть полезны для вас, особенно instance_dict(),
Вы можете попытаться сделать это таким образом.
for u in session.query(User).all():
print(u._asdict())
Он использует встроенный метод в объекте запроса, который возвращает диктонарный объект объекта запроса.
ссылки: https://docs.sqlalchemy.org/en/latest/orm/query.html
Я занимаюсь этим вопросом несколько минут. Ответ, отмеченный как правильный, не соответствует типу полей. Решение приходит из дикталхимии, добавляя некоторые интересные возможности. https://pythonhosted.org/dictalchemy/ Я только что протестировал и работает нормально.
Base = declarative_base(cls=DictableModel)
session.query(User).asdict()
{'id': 1, 'username': 'Gerald'}
session.query(User).asdict(exclude=['id'])
{'username': 'Gerald'}
class User(object):
def to_dict(self):
return dict([(k, getattr(self, k)) for k in self.__dict__.keys() if not k.startswith("_")])
Это должно работать.
Вы можете преобразовать объект sqlalchemy в словарь и вернуть его как json/dictionary.
Вспомогательные функции:
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)
Функция водителя:
def all_products():
all_products = Products.query.all()
return to_array(all_products)
Два пути:
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())
Чтобы завершить ответ @Anurag Uniyal, вот метод, который будет рекурсивно следовать отношениям:
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}
Как заявил OP, вызов инициализатора dict вызывает исключение с сообщением, что объект «Пользователь» не является итерируемым. Итак, реальный вопрос заключается в том, как сделать модель SQLAlchemy итерируемой?
Придется реализовать специальные методы
Мы можем реализовать наш собственный базовый класс, который реализует необходимые специальные методы и хранит список имен столбцов, к которым можно получить доступ по индексу, уменьшая временную сложность создания словаря до O (N). Это имеет дополнительное преимущество, заключающееся в том, что мы можем определить логику один раз и наследовать от нашего базового класса в любое время, когда мы хотим, чтобы наш класс модели был итерируемым.
class IterableBase(declarative_base()):
__abstract__ = True
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
self._keys = [c.name for c in self.__table__.columns]
self._dict = {c.name: getattr(self, c.name) for c in self.__table__.columns}
def __setattr__(self, name, value):
super().__setattr__(name, value)
if name not in ('_dict', '_keys', '_n') and '_dict' in self.__dict__:
self._dict[name] = value
def __iter__(self):
self._n = 0
return self
def __next__(self):
if self._n >= len(self._keys):
raise StopIteration
self._n += 1
key = self_keys[self._n-1]
return (key, self._dict[key])
Теперь класс User может напрямую наследоваться от нашего класса IterableBase.
class User(IterableBase):
__tablename__ = 'users'
id = Column(Integer, primary_key=True)
name = Column(String)
def __init__(self, name):
super().__init__(name=name)
Вы можете подтвердить, что вызов функции dict с экземпляром пользователя в качестве аргумента возвращает нужный словарь без «_sa_instance_state». Возможно, вы заметили
def main():
user1 = User('Bob')
print(dict(user1))
# outputs {'id': None, 'name': 'Bob'}
user1.id = 42
print(dict(user1))
# outputs {'id': 42, 'name': 'Bob'}
if __name__ == '__main__':
main()
После запроса базы данных с использованием следующего кода SQLAlchemy:
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
SQLALCHEMY_DATABASE_URL = 'sqlite:///./examples/sql_app.db'
engine = create_engine(SQLALCHEMY_DATABASE_URL, echo=True)
query = sqlalchemy.select(TABLE)
result = engine.execute(query).fetchall()
Вы можете использовать этот однострочный:
query_dict = [record._mapping for record in results]
С помощью этого кода вы также можете добавить к вашему запросу "фильтр" или "присоединиться" и эта работа!
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]
Ради всех и меня, вот как я использую это:
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]
Вот как это делает Elixir. Ценность этого решения состоит в том, что оно позволяет рекурсивно включать словарное представление отношений.
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
from copy import copy
def to_record(row):
record = copy(row.__dict__)
del record["_sa_instance_state"]
return record
Если не использовать копию, вы можете столкнуться с ошибками.
Мы можем получить список объектов в dict:
def queryset_to_dict(query_result):
query_columns = query_result[0].keys()
res = [list(ele) for ele in query_result]
dict_list = [dict(zip(query_columns, l)) for l in res]
return dict_list
query_result = db.session.query(LanguageMaster).all()
dictvalue=queryset_to_dict(query_result)
У меня есть вариант ответа Марко Мариани, выраженный в качестве декоратора. Основное отличие состоит в том, что он будет обрабатывать списки сущностей, а также безопасно игнорировать некоторые другие типы возвращаемых значений (что очень полезно при написании тестов с использованием 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