python - 将 SqlAlchemy orm 结果转换为 dict

Question

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

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

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)

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

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 方法不会自动扩展关系对象并将其转换为字典。

谁能建议我如何自动确定返回的用户对象是否有关系，如果它有一个关系对象，则将关系对象扩展为字典，等等。

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 查询。