simple user

auth.py

import http
import time

from core.settings import settings
from fastapi import HTTPException, Request
from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer
from jose import jwt


def decode_token(token: str) -> dict | None:
    try:
        decoded_token = jwt.decode(token, settings.authjwt_secret_key, algorithms=[settings.authjwt_algorithm])
        return decoded_token if decoded_token['exp'] >= time.time() else None
    except Exception:
        return None


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

    async def __call__(self, request: Request):
        credentials: HTTPAuthorizationCredentials = await super().__call__(request)
        if credentials:
            if not credentials.scheme == 'Bearer':
                raise HTTPException(status_code=http.HTTPStatus.UNAUTHORIZED, detail='Invalid authentication scheme.')
            decoded_token = self.parse_token(credentials.credentials)
            if not decoded_token:
                raise HTTPException(status_code=http.HTTPStatus.FORBIDDEN, detail='Invalid token or expired token.')
            return decoded_token
        raise HTTPException(status_code=http.HTTPStatus.FORBIDDEN, detail='Invalid authorization code.')

    @staticmethod
    def parse_token(jwt_token: str) -> dict | None:
        return decode_token(jwt_token)


security_jwt_local = JWTBearer()

routes.py

from typing import Annotated
from uuid import UUID

import grpc
from core.auth import security_jwt_local
from core.grpc import users_pb2, users_pb2_grpc
from core.settings import settings
from fastapi import APIRouter, Body, Depends
from google.protobuf.json_format import MessageToDict

router = APIRouter()


@router.get('/')
async def get_my_creds
    user: Annotated[dict, Depends(security_jwt_local)],
):
    # user - словарь с данными пользователя из токена, тут можно с ним работать, проверять доступ и прочее
    return user