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_super_and_project(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)) if user.id != 1: raise HTTPException(status_code=402, detail='该用户不拥有此权限') 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)) 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='该用户没有此接口的权限')