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 != 1:
        raise Exception("暂无权限")
    g.user_id = token_data.user_id
    g.user_name = token_data.user_name
    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 [1,2,3]:
        raise Exception("暂无权限")
    g.user_id = token_data.user_id
    g.user_name = token_data.user_name
    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.user_name = token_data.user_name
    g.project_id = token_data.project_id
    return token_data

async def verify_project_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 not token_data.role_id in [1,2,4]:
        raise Exception("暂无权限")
    g.user_id = token_data.user_id
    g.user_name = token_data.user_name
    g.project_id = token_data.project_id
    return token_data