auth.py

from functools import wraps

from async_fastapi_jwt_auth import AuthJWT
from fastapi import Depends, Request, status
from fastapi.security import HTTPBearer
from sqlalchemy.ext.asyncio import AsyncSession

from db.db import get_session
from helpers.auth_request import AuthRequest
from helpers.exceptions import AuthException
from models.user import UserRole
from schemas.user import UserInDb
from services.user_repository import users_crud


def roles_required(roles_list: list[UserRole]):
    def decorator(function):
        @wraps(function)
        async def wrapper(*args, **kwargs):
            user: UserInDb = kwargs.get('request').custom_user
            if not user or user.role not in [x.value for x in roles_list]:
                raise AuthException(
                    'This operation is forbidden for you', status_code=status.HTTP_403_FORBIDDEN,
                )
            return await function(*args, **kwargs)
        return wrapper
    return decorator


class JWTBearer(HTTPBearer):
    def __init__(self, auto_error: bool = True):
        super().__init__(auto_error=auto_error)

    async def __call__(self, request: Request, db: AsyncSession = Depends(get_session)) -> UserInDb | None:
        authorize = AuthJWT(req=request)
        await authorize.jwt_optional()
        user_id = await authorize.get_jwt_subject()
        if not user_id:
            return None
        user = await users_crud.get(db=db, id=user_id)
        return UserInDb.from_orm(user)


async def get_current_user_global(request: AuthRequest, user: AsyncSession = Depends(JWTBearer())):
    request.custom_user = user

auth_request.py

from fastapi import Request

from schemas.user import UserInDb


class AuthRequest(Request):
    custom_user: UserInDb

endpoints.py

import uuid

from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy.ext.asyncio import AsyncSession

from core.settings import settings
from db.db import get_session
from helpers.auth import roles_required
from helpers.auth_request import AuthRequest
from models.user import UserRole
from schemas.login_history import LoginHistoryResponse
from schemas.user import UserInDb, UserResponse, UserShort, UserUpdateRoleDto
from services.login_history_repository import history_crud
from services.user_repository import users_crud

router = APIRouter()


@router.get('/', response_model=list[UserShort])
@roles_required(roles_list=[UserRole.admin, UserRole.privileged_user])
async def read_users(
    *,
    request: AuthRequest,
    db: AsyncSession = Depends(get_session),
    skip: int = Query(0, description='Items to skip', ge=0),
    limit: int = Query(settings.pagination_limit, description='Items amount on page', ge=1),
) -> list[UserShort]:
    entities = await users_crud.get_multi(db=db, skip=skip, limit=limit)
    return entities

main.py

...

app.include_router(
    users_router, prefix='/api/v1/users', tags=['users'], dependencies=[Depends(get_current_user_global)],
)

...