Convert SqlAlchemy spelling result to dict

Question

How to convert the result of an orm object to an SQLAlchemy object in JSON format?

I am currently using sqlalchemy reflection to mirror tables from a database. Think of it, I have a User table and a table of addresses that I reflect in the database. A user object is one to one related to an address object. Below is the code reflecting the table from the database and using the mapper class to match the relationship.

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

Now when I query an object using sqlalchemy orm

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

Now that I want to convert the user object to a dict I use the method below.

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

However, the object_to_dict methods work fine and return a valid dic if the returned object is a user that is not related to another table. If the user object is related, the object_to_dict method does not automatically expand the relationship object and convert it to a dict.

Can someone suggest me how I can automatically determine if the returned user object has relationships and extends the relationship object to dict if it has one, etc. for any number of children.

Answer 1

You can use the mapper relationship property. The choice of code depends on how you want to match your data and what your relationship looks like. If you have a lot of recursive relationships, you can use the max_depth counter. The example below uses a set of relationships to prevent a recursive loop. You can completely eliminate recursion if you plan to go down in depth, but you said "and so on."

 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 

Also, keep in mind that there are performance issues. You can use parameters such as joinload or subqueryload to prevent an excessive number of SQL queries from executing.

Answer 2

Although the answer "doog adibies" was accepted, and I supported it because it was extremely useful, there are several important problems in the algorithm:

• Relationship sub-serialization stops at the first child (due to premature addition to " found ")
• It also serializes feedback relationships that are inaccessible in most cases (if you have a Father object with a Son attitude with backref configured, you will create an additional Father node for each son in it, with the same data as the main Father object!)

To fix these problems, I defined another set() to track unwanted feedback, and I migrated the tracking of visited children later in the code. I also intentionally renamed the variables to make it clearer (of course, IMO) what they represent and how the algorithm works, and replaced map() with a more understandable understanding of the dictionary.

Below is my actual working implementation, which was tested against nested objects from 4 dimensions (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