Convert SqlAlchemy orm result to dict

Question

How to convert SQLAlchemy orm object result to JSON format?

Currently I am using sqlalchemy reflection to reflect tables from the DB. Consider I have a User table and a Address table I am reflecting fro the DB. The user entity has one to one relationship with the address entity. Below is the code to reflect the table from the DB and use the mapper class to map 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 the object using sqlalchemy orm

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

Now, when I want to convert the user object to dict I use the below method

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 works fine and returns a valid dic object if the returned user object that did not have a relationship with another table. If the user object has a relationship the object_to_dict method doesn't auto-expand relations object and convert it to dict.

Could anyone suggest me how I could automatically determine if the returned user object has a relationship and expand the relationship object to a dict if it has one and so on for any number of child objects.

Answer 1

You can use the relationships property of the mapper. The code choices depend on how you want to map your data and how your relationships look. If you have a lot of recursive relationships, you may want to use a max_depth counter. My example below uses a set of relationships to prevent a recursive loop. You could eliminate the recursion entirely if you only plan to go down one in depth, but you did say "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, be aware that there are performance issues to consider. You may want to use options such as joinedload or subqueryload in order to prevent executing an excessive number of SQL queries.

Answer 2

Despite "doog adibies" answer has been accepted and I upvoted it since has been extremely helpful, there are a couple of notable issues in the algorithm:

1. The sub-serialization of relationships stops at the first child (because of the premature addition to "found")
2. It also serializes back relationships, which in most cases are not desiderable (if you have a Father object with a relationship to Son with a configured backref, you will generate an extra Father node for each son in it, with the same data that the main Father object already provides!)

To fix these issues, I defined another set() to track undesired back relationships and I moved the tracking of visited children later in the code. I also deliberately renamed variables in order to make more clear (of course IMO) what they represents and how the algorithm works and replaced the map() with a cleaner dictionary comprehension.

The following is my actual working implementation, which has been tested against nested objects of 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

Answer 3

Based on answers by "doog abides" and "daveoncode", with documentation and minor correction (as mentioned by "iuridiniz")

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

Answer 4

In case someone is using FastAPI and Sqlalchemy together then the following code can work.

You can use jsonable_encoder to covert model objects to dict.

from fastapi.encoders import jsonable_encoder
from models import Item

item = db.query(Item).filter(Item.id == some_id).first()

print(jsonable_encoder(item))

Source: https://onebite.dev/how-to-convert-sqlalchemy-object-to-json-in-fastapi/