datax-admin/app/routers/auth.py

import json
from typing import List
from fastapi import Depends
from fastapi import APIRouter
from fastapi.security import OAuth2PasswordRequestForm
from app import schemas
from app.utils.utils import encode_base64
from utils.sx_time import sxtimeit
from utils.sx_web import web_try
from configs.settings import config

super_admin_role = config.get('PERMISSIONS', 'super_admin_role')
project_admin_role = config.get('PERMISSIONS', 'project_admin_role')
special_project_id = config.get('PERMISSIONS', 'special_project_id')

router = APIRouter(
    prefix="/jpt/auth",
    tags=["auth-接口文档生成token"],
)


@router.post("/switch")
@web_try()
@sxtimeit
def switch_project(switch: schemas.SwitchProject):
    role_id = 0
    if switch.project_id == special_project_id and super_admin_role in switch.role_ids:
        role_id = 1
    elif switch.project_id == special_project_id and project_admin_role in switch.role_ids:
        role_id = 2
    elif switch.project_id == special_project_id:
        role_id = 3
    elif project_admin_role in switch.role_ids or super_admin_role in switch.role_ids:
        role_id = 4
    else:
        role_id = 5
    jupyter_download = False
    if super_admin_role in switch.role_ids:
        jupyter_download = True
    token_data = {"user_id": switch.user_id, "user_name": switch.user_name,
                  "project_id": switch.project_id, "role_id": role_id}
    token_data_str = json.dumps(token_data)
    access_token = encode_base64(token_data_str).replace('\n', '')
    return {"access_token": access_token, "token_type": "Bearer", "role_id": role_id, "jupyter_download": jupyter_download}


@router.post("/login", response_model=schemas.Token)
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user_id, project_id = form_data.username, form_data.password
    token_data = {"user_id": user_id, "user_name": "刘涛",
                  "project_id": project_id, "role_id": 1}
    token_data_str = json.dumps(token_data)
    access_token = encode_base64(token_data_str).replace('\n', '')
    return {"access_token": access_token, "token_type": "bearer"}