如何将 SQLAlchemy 行对象转换为 Python 字典？

Question

有没有一种简单的方法来迭代列名和值对？

我的 SQLAlchemy 版本是 0.5.6

这是我尝试使用dict(row)的示例代码：

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)

在我的系统输出上运行此代码：

Traceback (most recent call last):
  File "untitled-1.py", line 37, in <module>
    print dict(u)
TypeError: 'User' object is not iterable

Answer 1

您可以访问 SQLAlchemy 对象的内部 __dict__，如下所示：

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

Answer 2

我无法得到一个好的答案，所以我使用了这个：

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}

Answer 3

根据 cmets 中的@zzzeek：

请注意，这是现代版本的正确答案 SQLAlchemy，假设“行”是核心行对象，而不是 ORM 映射 实例。

for row in resultproxy:
    row_as_dict = dict(row)

Answer 4

在 SQLAlchemy v0.8 及更高版本中，使用inspection system。

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。

Answer 5

行有一个_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'}

Answer 6

正如@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'}

Answer 7

假设将以下函数添加到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 或您添加到对象中。 Reference

编辑：忘了说，这也适用于继承的列。

hybrid_property 扩展名

如果您还想包含 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 访问该属性，或者您只是不想显示它们。 Reference

Tipp all_orm_descriptors 也返回 hybrid_method 和 AssociationProxy 如果您还想包含它们。

其他答案的备注

每个基于 __dict__ 属性的答案（如 1、2 ）只返回对象的所有属性。这可能是您想要的更多属性。我很伤心，这包括 _sa_instance_state 或您在此对象上定义的任何其他属性。

基于dict() 函数的每个答案（如1、2）仅适用于session.execute() 返回的SQLalchemy 行对象，不适用于您定义的类与问题中的class User 类似。

基于row.__table__.columns 的solving answer 肯定不会工作。 row.__table__.columns 包含 SQL 数据库的列名。这些只能等于python对象的属性名。如果没有，你会得到一个AttributeError。 对于基于class_mapper(obj.__class__).mapped_table.c 的答案（如1、2）是一样的。

Answer 8

老问题，但由于这是 Google 中“sqlalchemy row to dict”的第一个结果，因此值得更好的回答。

SqlAlchemy 返回的 RowProxy 对象具有 items() 方法： http://docs.sqlalchemy.org/en/latest/core/connections.html#sqlalchemy.engine.RowProxy.items

它只是返回一个 (key, value) 元组的列表。因此，可以使用以下命令将一行转换为 dict：

在 Python

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]

Answer 9

根据@balki 的回答，从 SQLAlchemy 0.8 开始，您可以使用_asdict()，可用于KeyedTuple 对象。这为原始问题提供了一个非常简单的答案。只是，在您的示例中将最后两行（for 循环）更改为这一行：

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

这是有效的，因为在上面的代码中u 是一个类型为KeyedTuple 的对象，因为.all() 返回KeyedTuple 的列表。因此它有 _asdict() 方法，它很好地将 u 作为字典返回。

WRT @STB 的答案：AFAIK，.all() 返回的任何内容都是KeypedTuple 的列表。因此，无论是否指定列，只要您将.all() 的结果应用于Query 对象，上述方法都有效。

Answer 10

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)

Answer 11

参考Alex Brasetvik's Answer，一行代码即可解决问题

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

在 Alex Brasetvik's Answer 的评论部分，SQLAlchemy 的创建者 zzzeek 表示这是解决问题的“正确方法”。

Answer 12

一个非常简单的解决方案：row._asdict()。

• sqlalchemy.engine.Row._asdict() (v1.4)
• sqlalchemy.util.KeyedTuple._asdict() (v1.3)

> data = session.query(Table).all()
> [row._asdict() for row in data]

Answer 13

您正在迭代的表达式计算为模型列表对象，而不是行。所以以下是它们的正确用法：

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

您真的需要将它们转换为字典吗？当然，有很多方法，但是你不需要 SQLAlchemy 的 ORM 部分：

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

更新：查看sqlalchemy.orm.attributes 模块。它有一组处理对象状态的函数，可能对你有用，尤其是instance_dict()。

Answer 14

我找到这篇文章是因为我正在寻找一种将 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()))

Answer 15

你可以试试这样。

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

它使用查询对象中的内置方法返回查询对象的字典对象。

参考：https://docs.sqlalchemy.org/en/latest/orm/query.html

Answer 16

使用 python 3.8+，我们可以使用 dataclass 以及它附带的 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 装饰器，并用其类型注释每一列（name: str = Column(String) 行的: str 部分）。

还要注意，由于email 没有注释，所以它不包含在query_result_dict 中。

Answer 17

使用 sqlalchemy 1.4

session.execute(select(User.id, User.username)).mappings().all()
>> [{'id': 1, 'username': 'Bob'}, {'id': 2, 'username': 'Alice'}]

Answer 18

class User(object):
    def to_dict(self):
        return dict([(k, getattr(self, k)) for k in self.__dict__.keys() if not k.startswith("_")])

应该可以的。

Answer 19

您可以像这样将 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)

Answer 20

我刚刚处理了这个问题几分钟。 标记为正确的答案不考虑字段的类型。 解决方案来自 dictalchemy 添加一些有趣的功能。 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'}

Answer 21

两种方式：

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 22

这就是 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

Answer 23

使用此代码，您还可以将“过滤器”或“加入”添加到您的查询中！

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 24

我对 Marco Mariani 的回答有一个变体，表示为装饰者。主要区别在于它将处理实体的列表，以及安全地忽略一些其他类型的返回值（这在使用模拟编写测试时非常有用）：

@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 25

为了大家和我自己，我是这样使用它的：

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 26

要完成@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}

Answer 27

我们可以得到一个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)

Answer 28

from copy import copy

def to_record(row):
    record = copy(row.__dict__)
    del record["_sa_instance_state"]
    return record

如果不使用副本，您可能会遇到错误。

Answer 29

我是一名刚入门的 Python 程序员，在使用 Joined 表获取 JSON 时遇到了问题。使用此处答案中的信息，我构建了一个函数以向 JSON 返回合理的结果，其中包含表名，避免了别名或字段冲突。

只需传递会话查询的结果：

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

json = sqlAl2json(测试)

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)

Answer 30

如果您的模型表列不是 equie mysql 列。

如：

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

需要使用：

 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}

如果你使用这种方式，你可以得到 modify_time 和 create_time 都是 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}

因为类属性名不等于mysql中的列存储