I am building a RBAC web app with FastAPI. This led me to model implementation below. Based on it, I wrote some tests to validate the implementation. Among others, I required to get permissions of an user based on current instance. Since model instance would be already available as soon I call the method, I thought, it would be straightforward to call required method to obtain the permissions, like a regular class instance. It did not work, but raised error sqlalchemy.exc.MissingGreenlet: greenlet_spawn has not been called; can't call await_only() here. Was IO attempted in an unexpected place?. ChatGPT suggested me to perform operation as on implementation, which worked, but made me a bit sad on limitations of the library for async operations. I step further would be to use a more adequate library SQLModel (or even this async-SQLModel) as a substitute, but I feel not so comfortable about that. Had you the same experience with async engine so far?

In case you are still curious about the web-app, it is available at URL https://github.com/brunolnetto/fastapi-auth-mvp/tree/main.

from sqlalchemy import (
    Column, String, Boolean, DateTime, UUID,
    ForeignKey, Table
)

from typing import Tuple
from sqlalchemy.sql import select
from sqlalchemy.orm import relationship, joinedload
from sqlalchemy.dialects.postgresql import JSONB
from uuid import uuid4

from datetime import datetime, timezone
from typing import Any

from . import Base


# Association tables for many-to-many relationships
users_roles_association = Table(
    'users_x_roles', Base.metadata,
    Column('user_id', UUID, ForeignKey('users.user_id', ondelete='cascade')),
    Column('role_id', UUID, ForeignKey('roles.role_id'))
)


roles_permissions_association = Table(
    'roles_x_permissions', Base.metadata,
    Column('role_id', UUID, ForeignKey('roles.role_id', ondelete='cascade')),
    Column('perm_id', UUID, ForeignKey('permissions.perm_id'))
)


class User(Base):
    __tablename__ = "users"

    user_id = Column(UUID, primary_key=True, index=True, default=uuid4)
    user_created_at = Column(DateTime(timezone=True), default=datetime.now(timezone.utc))
    user_updated_at = Column(DateTime(timezone=True), default=None, onupdate=datetime.now(timezone.utc))
    user_last_login = Column(DateTime, default=None, nullable=True)
    user_username = Column(String, unique=True, index=True, nullable=False)
    user_email = Column(String, unique=True, index=True, nullable=False)
    user_hashed_password = Column(String, nullable=False)
    user_is_active = Column(Boolean, default=True)
    user_access_token = Column(String, nullable=True)
    user_refresh_token = Column(String, nullable=True)

    user_roles = relationship(
        'Role', secondary=users_roles_association, back_populates='role_users', cascade="all"
    )
    user_request_logs = relationship('RequestLog', back_populates='relo_user')

    def has_roles(self, allowed_roles: Tuple[str]):
        user_roles_set = {
            role.role_name for role in self.user_roles
        }
        
        allowed_roles_set = set(allowed_roles)
        intersection_roles = user_roles_set.intersection(allowed_roles_set)
        
        return len(intersection_roles) == len(allowed_roles_set)

    def has_permissions(self, allowed_permissions: Tuple[str]):
        user_permissions_set = {
            perm.perm_name for role in self.user_roles for perm in role.role_permissions
        }
        
        allowed_permissions_set = set(allowed_permissions)
        return not user_permissions_set.isdisjoint(allowed_permissions_set)

    async def get_permissions(self, session):
        result = await session.execute(
            select(User)
            .options(
                joinedload(User.user_roles).joinedload(Role.role_permissions)
            )
            .filter(User.user_id == self.user_id)
        )
        user = result.unique().fetchall()[0][0]
        
        permissions = {
            perm.perm_name
            for role in user.user_roles
            for perm in role.role_permissions
        }
        return permissions

    def __repr__(self):
        return f"User({self.user_username})"

    def __str__(self):
        return self.__repr__()

    def __eq__(self, other: Any) -> bool:
        if isinstance(other, User):
            equal_username=self.user_username == other.user_username
            equal_email=self.user_email == other.user_email
            
            return equal_username or equal_email
        else:
            return False

class Role(Base):
    __tablename__ = 'roles'
    role_id = Column(UUID, primary_key=True)
    role_name = Column(String, unique=True, nullable=False)
    role_permissions = relationship(
        'Permission', secondary=roles_permissions_association, back_populates='perm_roles'
    )
    role_users = relationship(
        'User', secondary=users_roles_association,  back_populates='user_roles'
    )

    def __repr__(self):
        permissions = ', '.join([perm.perm_name for perm in self.role_permissions])
        return f"Role({permissions})"

class Permission(Base):
    __tablename__ = 'permissions'
    perm_id = Column(UUID, primary_key=True)
    perm_name = Column(String, unique=True, nullable=False)
    perm_roles = relationship(
        'Role', secondary=roles_permissions_association, back_populates='role_permissions'
    )

    def __repr__(self):
        return f"Permission({self.perm_name})"

I thank you sincerely for your help.

Best regards.