datax-admin/app/common/security/auth.py

import json
from fastapi import Depends
from fastapi.security import OAuth2PasswordBearer
from app import schemas
from app.utils.utils import decode_base64
from configs.globals import g


reuseable_oauth = OAuth2PasswordBearer(
    tokenUrl="/jpt/auth/login",
    scheme_name="JWT"
)


async def verify_super_admin(token: str = Depends(reuseable_oauth)) -> schemas.TokenData:
    token_data_str = decode_base64(token)
    token_data_dict = json.loads(token_data_str)
    token_data = schemas.TokenData(**token_data_dict)
    if token_data.role_id != '726a51e45b4d11edbb4809c4df301a':
        raise Exception("暂无权限")
    g.user_id = token_data.user_id
    g.project_id = token_data.project_id
    return token_data


async def verify_special(token: str = Depends(reuseable_oauth)) -> schemas.TokenData:
    token_data_str = decode_base64(token)
    token_data_dict = json.loads(token_data_str)
    token_data = schemas.TokenData(**token_data_dict)
    if not token_data.role_id in ['726a51e45b4d11edbb4809c4df301a','9ff183445b4d11ed87db29f50d093a']:
        raise Exception("暂无权限")
    g.user_id = token_data.user_id
    g.project_id = token_data.project_id
    return token_data


async def verify_users(token: str = Depends(reuseable_oauth)) -> schemas.TokenData:
    token_data_str = decode_base64(token)
    token_data_dict = json.loads(token_data_str)
    token_data = schemas.TokenData(**token_data_dict)
    g.user_id = token_data.user_id
    g.project_id = token_data.project_id
    return token_data