Convertir SqlAlchemy orm resultado a dict

¿Cómo convertir el resultado del objeto orm SQLAlchemy al formato JSON?

Actualmente estoy usando la reflexión sqlalchemy para reflejar las tablas de la base de datos. Considerar que tengo una tabla de usuario y una tabla de direcciones que estoy reflejando para la base de datos. La entidad de usuario tiene una relación uno a uno con la entidad de dirección. A continuación se muestra el código para reflejar la tabla de la base de datos y usar la clase de asignador para asignar la relación.

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

Ahora cuando pregunto el objeto usando sqlalchemy orm

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

Ahora, cuando quiero convertir el objeto de usuario a dictado, uso el siguiente método

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

Sin embargo, los métodos object_to_dict funcionan bien y devuelven un objeto dic válido si el objeto de usuario devuelto no tenía una relación con otra tabla. Si el objeto de usuario tiene una relación, el método object_to_dict no expande automáticamente el objeto de relaciones y lo convierte en dict.

¿Podría alguien sugerirme cómo podría determinar automáticamente si el objeto de usuario devuelto tiene una relación y expandir el objeto de relación a un dictado si tiene uno y así sucesivamente para cualquier número de objetos secundarios?

Puede utilizar la propiedad de relaciones del asignador. Las opciones de código dependen de cómo desea asignar sus datos y cómo se ven sus relaciones. Si tiene muchas relaciones recursivas, puede usar un contador max_depth. Mi ejemplo a continuación utiliza un conjunto de relaciones para evitar un bucle recursivo. Podría eliminar la recursión por completo si solo planea profundizar una, pero dijo “y así sucesivamente”.

 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 

Además, tenga en cuenta que hay problemas de rendimiento a considerar. Es posible que desee utilizar opciones como la carga conjunta o la subconsulta para evitar la ejecución de un número excesivo de consultas SQL.

A pesar de que “doog adibies” la respuesta ha sido aceptada y la he subestimado ya que ha sido extremadamente útil, hay un par de problemas notables en el algoritmo:

  1. La subserie de relaciones se detiene en el primer hijo (debido a la adición prematura de ” found “)
  2. También serializa las relaciones de retroceso, que en la mayoría de los casos no son desagradables (si tiene un objeto de Father con una relación de Son con un backref configurado, generará un nodo de Father adicional para cada hijo en él, con los mismos datos que el principal Father objeto del Father ya proporciona!)

Para solucionar estos problemas, definí otro set() para rastrear las relaciones de pareja no deseadas y moví el rastreo de los niños visitados más adelante en el código. También cambié deliberadamente el nombre de las variables para dejar más claro (por supuesto, la OMI) lo que representan y cómo funciona el algoritmo, y reemplacé el map() con una comprensión de diccionario más limpia.

La siguiente es mi implementación de trabajo real, que ha sido probada contra objetos nesteds de 4 dimensiones (Usuario -> UsuarioProyecto -> UsuarioProjectEntity -> UsuarioProjectEntityField):

 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