datax-admin/app/common/decorators.py

from typing import Optional
from fastapi import FastAPI, Header, HTTPException, Depends, Request
from sqlalchemy.orm import Session
from app import crud, get_db, get_page, schemas
from configs.globals import g

async def verify_user(request: Request, user_token: str = Header(), db: Session = Depends(get_db)):
    try:
        user = crud.verify_user_token(db, user_token)
        g.user_id = user.id
        g.user_name = user.name
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))

async def verify_super_admin(request: Request, user_token: str = Header(), db: Session = Depends(get_db)):
    try:
        user = crud.verify_user_token(db, user_token)
        g.user_id = user.id
        g.user_name = user.name
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))
    if user.id != 1:
        raise HTTPException(status_code=402, detail='该用户不拥有此权限')

async def verify_all(request: Request, user_token: str = Header(), item_token: str = Header(), db: Session = Depends(get_db)):
    try:
        user = crud.verify_user_token(db, user_token)
        g.user_id = user.id
        g.user_name = user.name
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))
    try:
        project = crud.verify_item_token(db, item_token)
        g.project_id = project.id
    except Exception as e:
        raise HTTPException(status_code=403, detail=str(e))
    relations = crud.get_relations_by_user(db,user.id)
    project_role = {r.project_id:r.role_id for r in relations}
    if not project.id in project_role.keys():
        raise HTTPException(status_code=402, detail='该用户不属于此项目')
    role = crud.get_role_info(db,project_role[project.id])

async def verify_special(request: Request, user_token: str = Header(), item_token: str = Header(), db: Session = Depends(get_db)):
    try:
        user = crud.verify_user_token(db, user_token)
        g.user_id = user.id
        g.user_name = user.name
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))
    try:
        project = crud.verify_item_token(db, item_token)
        g.project_id = project.id
    except Exception as e:
        raise HTTPException(status_code=403, detail=str(e))
    relations = crud.get_relations_by_user(db,user.id)
    project_role = {r.project_id:r.role_id for r in relations}
    if not project.id in project_role.keys():
        raise HTTPException(status_code=402, detail='该用户不属于此项目')
    role = crud.get_role_info(db,project_role[project.id])
    if not role.code in ['726a51e45b4d11edbb4809c4df301a','9ff183445b4d11ed87db29f50d093a','eef984e65b4d11ed8cc491f9c82401']:
        raise HTTPException(status_code=402, detail='该用户没有此接口的权限')

async def verify_admin(request: Request, user_token: str = Header(), item_token: str = Header(), db: Session = Depends(get_db)):
    try:
        user = crud.verify_user_token(db, user_token)
        g.user_id = user.id
        g.user_name = user.name
    except Exception as e:
        raise HTTPException(status_code=401, detail=str(e))
    try:
        project = crud.verify_item_token(db, item_token)
        g.project_id = project.id
    except Exception as e:
        raise HTTPException(status_code=403, detail=str(e))
    relations = crud.get_relations_by_user(db,user.id)
    project_role = {r.project_id:r.role_id for r in relations}
    if not project.id in project_role.keys():
        raise HTTPException(status_code=402, detail='该用户不属于此项目')
    role = crud.get_role_info(db,project_role[project.id])
    if not role.code in ['726a51e45b4d11edbb4809c4df301a','9ff183445b4d11ed87db29f50d093a','026bd8bc5b4e11ed857e6b5ec5c8d6']:
        raise HTTPException(status_code=402, detail='该用户没有此接口的权限')