Function is_user_superadmin

apps/api/src/security/superadmin.py:11–22

Check if a user is a superadmin by querying the database directly.

(user_id: int, db_session: AsyncSession)

async def is_user_superadmin(user_id: int, db_session: AsyncSession) -> bool:
    """Check if a user is a superadmin by querying the database directly."""
    _cache = getattr(db_session, '_superadmin_cache', None)
    if _cache is None:
        db_session._superadmin_cache = {}
        _cache = db_session._superadmin_cache
    if user_id in _cache:
        return _cache[user_id]
    result = (await db_session.execute(select(User.is_superadmin).where(User.id == user_id))).scalars().first()
    value = bool(result)
    _cache[user_id] = value
    return value


async def _get_current_user_lazy(request: Request, db_session: AsyncSession = Depends(get_db_session)):