Role based row filtering: Advanced SQLAlchemy Techniques

In the article Mastering Soft Delete: Advanced SQLAlchemy Techniques, we delved into techniques for implementing global filters based on is_deleted flag. Building on this foundational knowledge, we now advance to applying global filters dynamically based on user roles and permissions.

we will explore how to implement role-based filtering strategies using SQLAlchemy. We’ll look at how to dynamically adjust query results based on the user’s role, allowing for a more granular and controlled approach to data access. This technique is particularly useful in applications with varying access levels and permissions.

We will use JWT tokens to handle user roles, which can be free_user, paid_user, or admin. Our database model will be extended with two additional columns: is_visible and is_paid, alongside the existing is_deleted column. The goal is to return only visible movies to all users, and restrict access to paid movies to paid_user and admin roles.

Here’s how we’ll approach the implementation:

Define the Models and Functions

class VisibilityMixin:
    is_deleted = Column(Boolean, server_default="0")
    is_visible = Column(Boolean, server_default="1")
    is_paid = Column(Boolean, server_default="0")

class Movie(DeclarativeBase, VisibilityMixin):
    __tablename__ = "movies"
    id = Column(Integer, primary_key=True)
    name = Column(String(255), nullable=False, index=True)
    mp4_url = Column(String(200))

oauth2_scheme = OAuth2AuthorizationCodeBearer(
    authorizationUrl="/login",
    tokenUrl="/refresh-token",
)
class Role(Enum):
    FREE_USER = "FREE_USER"
    PAID_USER = "PAID_USER"
    ADMIN = "admin"

def decode_token(token: str) -> dict:
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except jwt.PyJWTError as e:
        raise HTTPException(
            status_code=HTTPStatus.UNAUTHORIZED,
            detail="Access token has expired",
            headers={"WWW-Authenticate": "Bearer"},
        )
def get_role(token: str = Depends(oauth2_scheme)):
    token_data = decode_token(token)
    return Role(token_data.get("roles", "FREE_USER")),

Now, we can integrate the get_role function as a FastAPI dependency. This allows us to automatically set the user's role every time we fetch a database session.

We modify the get_db function to include the user’s role in the database session. This ensures that every time a session is created, it includes the current user's role for filtering purposes.With the user’s role now set in the session, the do_orm_execute event listener dynamically applies filters based on the role. This ensures that queries are automatically adjusted to return data according to the user’s permissions.

engine = create_engine(DATABASE_URL, pool_recycle=3600)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@event.listens_for(SessionLocal, "do_orm_execute")
def _add_filtering_criteria(execute_state: ORMExecuteState):
    skip_filter = execute_state.execution_options.get("skip_visibility_filter", False)
    if execute_state.is_select and not skip_filter:
        role: Role = execute_state.session.info.get("role", Role.FREE_USER)

        def admin_filter(cls):
            return cls.is_deleted.is_(False)

        def free_user_filter(cls):
            return cls.is_deleted.is_(False) & cls.is_visible.is_(True) & cls.is_paid.is_(False)

        def paid_user_filter(cls):
            return cls.is_deleted.is_(False) & cls.is_visible.is_(True)
        filters = {
            Role.FREE_USER: free_user_filter,
            Role.PAID_USER: paid_user_filter,
            Role.ADMIN: admin_filter,
        }

        execute_state.statement = execute_state.statement.options(
            with_loader_criteria(VisibilityMixin, filters[role], include_aliases=True)
        )


def get_db(
    role: Role = Depends(get_role),
):
    db: Session = SessionLocal()
    db.info["role"] = role
    try:
        yield db
    finally:
        db.close()



@app.get("/movie")
async def get_movies(
    db: Session = Depends(get_db),
    role: Role = Depends(get_role),
):
    return db.query(Movie).all()

When defining your API endpoints, you can now simply depend on the get_db function, and the appropriate filters will be applied automatically based on the role.

Conclusion

In this article, we’ve explored how to leverage role-based row filtering in SQLAlchemy to enhance data security and efficiency. Building on our previous discussion of soft deletes, we’ve implemented dynamic filters based on user roles, which allows for more granular control over data access.

As you incorporate these techniques into your applications, you’ll find that managing access levels and permissions becomes more intuitive and maintainable. By dynamically adjusting query results based on roles, you can create more robust and user-centric data access strategies.