0

I have two models - User and UserRole. I want to be able to get the user role together with the related user model.

I managed to do it like this:

@staticmethod
async def get_users(session: AsyncSession):
    users = await session.execute(
        select(User, UserRole).join(UserRole, UserRole.role_id == User.role_id)
    )

    result = users.all()
    users = []
    for user in result:
        users.append({**user.User.dict(), "role": user.UserRole})

    return users

My models look like this:

class User(SQLModel, table=True):
    user_id: Optional[int] = Field(default=None, primary_key=True, nullable=False)
    username: str
    role_id: Optional[int] = Field(default=3, foreign_key="user_role.role_id")

class UserRole(SQLModel, table=True):
    __tablename__ = "user_role"
    role_id: Optional[int] = Field(default=None, primary_key=True, nullable=False)
    role_name: str

The code from get_users returns the following:

[{
    "user_id": 1,
    "username": "admin",
    "role_id": 1,
    "role": {
        "role_id": 1,
        "role_name": "admin"
    }
},
{
    "user_id": 2,
    "username": "user",
    "role_id": 2,
    "role": {
        "role_id": 2,
        "role_name": "user"
    }
}]

The code above works fine but the thing is that I'm not sure this is the best thing to do and also I'm not using any Relationship field inside the models, as mentioned in the SQLModel documentation. Is that OK? Why do I need to use it, if it works without the Relationship? Also, is there a shorter way to do this without looping through the result and append to it?

Daniil Fajnberg
  • 12,753
  • 2
  • 10
  • 41
Or Nakash
  • 1,345
  • 1
  • 9
  • 22
  • [This Q&A](https://stackoverflow.com/q/74252768/5320906), though a debugging question, shows how you might perform the join implicitly using a relationship. To build the JSON you could consider libraries like marshmallow or pydantic. – snakecharmerb Nov 01 '22 at 09:46

1 Answers1

2

Relationship attributes do in fact make these sorts of operations much simpler and more readable IMHO. One thing you can configure about your relationships is their loading behavior with regards to database queries. Since SQLModel is essentially a wrapper around SQLAlchemy, configuration is very similar.

Currently, the Relationship function from SQLModel expects you to use the sa_relationship-parameters to pass arguments to SQLAlchemy's relationship constructor. The relevant parameter here is lazy; the meaning of the different loading behaviors is documented in more detail here. Thus, you can for example define your relationships like this:

from typing import Optional

from sqlmodel import (
    Field,
    Relationship,
    Session,
    SQLModel,
    create_engine,
    select,
)

class User(SQLModel, table=True):
    user_id: Optional[int] = Field(default=None, primary_key=True)
    username: str

    role_id: Optional[int] = Field(
        default=None,
        foreign_key="user_role.role_id",
    )
    role: Optional["UserRole"] = Relationship(
        back_populates="users",
        sa_relationship_kwargs=dict(lazy="selectin"),  # depends on your needs
    )

class UserRole(SQLModel, table=True):
    __tablename__ = "user_role"
    role_id: Optional[int] = Field(default=None, primary_key=True)
    role_name: str

    users: list[User] = Relationship(back_populates="role")

# more code below ...

Let's test that the ORM behaves as we would expect it to: (I use non-async code and an in-memory SQLite database for simplicity.)

def test() -> None:
    # Initialize database & session:
    engine = create_engine("sqlite:///", echo=True)
    SQLModel.metadata.drop_all(engine)
    SQLModel.metadata.create_all(engine)
    session = Session(engine)

    # Create the test objects:
    role = UserRole(role_name="admin")
    User(username="Foo", role=role)
    User(username="Bar", role=role)
    session.add(role)
    session.commit()

    # Get all users:
    statement = select(User)
    users = session.execute(statement).scalars()

    # Assemble dictionaries:
    user_dicts = [
        user.dict(exclude={"role_id"}) | {"role": user.role.dict()}
        for user in users
    ]
    import json
    print(json.dumps(user_dicts, indent=4))

if __name__ == '__main__':
    test()

Here is the output of that last print statement:

[
    {
        "username": "Foo",
        "user_id": 1,
        "role": {
            "role_id": 1,
            "role_name": "admin"
        }
    },
    {
        "username": "Bar",
        "user_id": 2,
        "role": {
            "role_id": 1,
            "role_name": "admin"
        }
    }
]

If we take a look at the SQL log output from our engine, we can see the following entries following the table creation statements:

...
SELECT user.user_id, user.username, user.role_id FROM user
...
SELECT user_role.role_id AS user_role_role_id, user_role.role_name AS user_role_role_name 
FROM user_role 
WHERE user_role.role_id IN (?)
(1,)

These SQL statements are in accordance to what we defined to be our lazy loading behavior. That second SELECT query is issued only at the moment, we attempt to access the user.role attribute the first time in our user_dicts list comprehension loop. There was no need to explicitly use join in our code.

A few other things to note:

  • We would still not need to use an explicit join, even if we hadn't explicitly defined the loading behavior in our relationship, but the number of issued SQL queries would be higher, if we had more than one role.
  • We use scalars to get an iterable of User instances immediately; this makes working with them more convenient.
  • Currently, by design, SQLModel excludes relationship attributes from methods such as dict or json because you'd need a mechanism for avoid infinite recursion otherwise. I think this may change in the future. Right now, we need to add that role.dict() ourselves.
  • I excluded role_id from the user dictionary, because it appears inside the role dictionary.
  • Dictionaries can be merged conveniently using | since Python 3.9 (see docs).
  • When primary_key is set to True on a field/column, nullable will be set to False by default, so you don't need to specify it.

If you use lazy=joined on your relationships, the related rows are queried eagerly right away in the first SELECT. That means you'll immediately have all the Role objects on your users. But if you need the dictionary representation, as I said, you'll still have to merge them manually.


PS: Model parsing and inheritance

I just remembered that you can get around the manual merging of dictionaries, if you leverage model inheritance properly. This is documented in the SQLModel examples with FastAPI.

You can define a non-table base model for users. Then you create one table-model that inherits from it looking essentially just like your model looked already and another non-table model that has regular (i.e. non-relationship) fields representing the related model. You can then construct that last model from instances of the table-model with e.g. from_orm.

Here are the changes to my example code above:

...

class UserBase(SQLModel):
    username: str

class User(UserBase, table=True):
    user_id: Optional[int] = Field(default=None, primary_key=True)

    role_id: Optional[int] = Field(
        default=None,
        foreign_key="user_role.role_id",
    )
    role: Optional["UserRole"] = Relationship(
        back_populates="users",
        sa_relationship_kwargs=dict(lazy="selectin"),  # depends on your needs
    )

class UserRole(SQLModel, table=True):
    ...

class UserWithRole(UserBase):
    user_id: int
    role: UserRole

...

def test() -> None:
    ...
    # Get all users:
    statement = select(User)
    users = session.execute(statement).scalars()

    # Assemble dictionaries:
    user_dicts = [UserWithRole.from_orm(user).dict() for user in users]

The output would be the same. This approach really shines, when your schemas/models become more and more complex.

If you wanted, you could even define a container model for multiple instances of UserWithRole and do the following in addition:

...

class UsersWithRole(SQLModel):
    users: list[UserWithRole]

...

def test() -> None:
    ...
    # Get all users:
    statement = select(User)
    users = list(session.execute(statement).scalars())

    # Assemble dictionaries:
    user_dicts = UsersWithRole.parse_obj({"users": users}).dict()["users"]
    ...

I think you get the idea.

Daniil Fajnberg
  • 12,753
  • 2
  • 10
  • 41