将 SqlAlchemy orm 结果转换为 dict

Question

如何将 SQLAlchemy orm 对象结果转换为 JSON 格式？

目前我正在使用 sqlalchemy 反射来反映数据库中的表。 考虑我有一个用户表和一个地址表，我正在从数据库中反映出来。 用户实体与地址实体具有一对一的关系。 下面是从数据库中反映表并使用映射器类映射关系的代码。

from sqlalchemy import Table
from sqlalchemy.orm import mapper, relationship
user_reflection = Table('user', metadata, autoload=True, autoload_with=engine)
class User(object):
    def __init__(self, id, name, dob):
        self.id = id
        self.name = name
        self.dob = dob
address_reflection = Table('address', metadata, autoload=True, autoload_with=engine)
mapper(User,
       user_reflection,
       properties={
           'address': relationship(SourceAddress, uselist=False)
       }
)

现在当我使用 sqlalchemy orm 查询对象时

user = session.query(User).first()
user_dict = object_to_dict(user)

现在，当我想将用户对象转换为 dict 我使用下面的方法

def object_to_dict(obj):
    columns = [column.key for column in class_mapper(obj.__class__).columns]
    get_key_value = lambda c: (c, getattr(obj, c).isoformat()) if isinstance(getattr(obj, c), datetime) else (c, getattr(obj, c))
    return dict(map(get_key_value, columns))

但是，如果返回的用户对象与另一个表没有关系，则 object_to_dict 方法可以正常工作并返回有效的 dic 对象。 如果用户对象具有关系，则 object_to_dict 方法不会自动扩展关系对象并将其转换为 dict。

任何人都可以建议我如何自动确定返回的用户对象是否具有关系，并将关系对象扩展为一个字典，如果它有一个，等等对于任意数量的子对象。

Answer 1

您可以使用映射器的关系属性。代码选择取决于您希望如何映射数据以及关系的外观。如果您有很多递归关系，您可能需要使用 max_depth 计数器。我下面的示例使用一组关系来防止递归循环。如果你只打算深入一个，你可以完全消除递归，但你确实说了“等等”。

def object_to_dict(obj, found=None):
    if found is None:
        found = set()
    mapper = class_mapper(obj.__class__)
    columns = [column.key for column in mapper.columns]
    get_key_value = lambda c: (c, getattr(obj, c).isoformat()) if isinstance(getattr(obj, c), datetime) else (c, getattr(obj, c))
    out = dict(map(get_key_value, columns))
    for name, relation in mapper.relationships.items():
        if relation not in found:
            found.add(relation)
            related_obj = getattr(obj, name)
            if related_obj is not None:
                if relation.uselist:
                    out[name] = [object_to_dict(child, found) for child in related_obj]
                else:
                    out[name] = object_to_dict(related_obj, found)
    return out

另外，请注意还有性能问题需要考虑。您可能希望使用joinedload 或subqueryload 等选项来防止执行过多的SQL 查询。

Answer 2

尽管“doog adibies”的答案已被接受并且我赞成它，因为它非常有帮助，但算法中有几个值得注意的问题：

1. 关系的子序列化在第一个孩子处停止（因为过早添加到“found”）
2. 它还会序列化反向关系，这在大多数情况下是不可取的（如果您有一个与 Son 的关系的 Father 对象并配置了 backref，您将为每个儿子在其中，具有与主要 Father 对象已经提供的相同数据！）

为了解决这些问题，我定义了另一个 set() 来跟踪不受欢迎的反向关系，并在代码的后面部分移动了对访问过的孩子的跟踪。我还特意重命名了变量，以便更清楚地（当然是 IMO）它们代表什么以及算法如何工作，并用更清晰的字典理解替换了 map()。

以下是我的实际工作实现，已针对 4 维嵌套对象（User -> UserProject -> UserProjectEntity -> UserProjectEntityField）进行了测试：

def model_to_dict(obj, visited_children=None, back_relationships=None):
    if visited_children is None:
        visited_children = set()
    if back_relationships is None:
        back_relationships = set()
    serialized_data = {c.key: getattr(obj, c.key) for c in obj.__table__.columns}
    relationships = class_mapper(obj.__class__).relationships
    visitable_relationships = [(name, rel) for name, rel in relationships.items() if name not in back_relationships]
    for name, relation in visitable_relationships:
        if relation.backref:
            back_relationships.add(relation.backref)
        relationship_children = getattr(obj, name)
        if relationship_children is not None:
            if relation.uselist:
                children = []
                for child in [c for c in relationship_children if c not in visited_children]:
                    visited_children.add(child)
                    children.append(model_to_dict(child, visited_children, back_relationships))
                serialized_data[name] = children
            else:
                serialized_data[name] = model_to_dict(relationship_children, visited_children, back_relationships)
    return serialized_data

Answer 3

基于“doog abides”和“daveoncode”的回答，并附有文档和较小的更正（如“iuridiniz”所述）

https://gist.github.com/hrishikeshrt/abb610743c394ce140196498b9c4ff0b