| 12345678910111213141516171819202122232425262728293031323334353637383940 |
- from enum import IntEnum
- from fastapi import Depends, HTTPException, status
- from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
- from app.core.biz_exception import NotFoundError, PermissionDeniedError, BizLogicError
- from app.core.config import settings
- from sqlalchemy.orm import Session
- from app.core.database import get_db
- from app.services.session_service import SessionService
- security = HTTPBearer()
- class RoleLevel(IntEnum):
- user = 10
- admin = 100
- ROLE_LEVEL_MAP = {
- "user": 10,
- "admin": 100,
- }
- def require_min_role(min_role: RoleLevel):
- def checker(user=Depends(get_current_user)):
- current_level = ROLE_LEVEL_MAP.get(user.role, 0)
- if current_level < min_role:
- raise PermissionDeniedError("Permission denied")
- return user
- return checker
- def get_current_user(
- credentials: HTTPAuthorizationCredentials = Depends(security),
- db: Session = Depends(get_db)
- ):
- token = credentials.credentials
- user = SessionService().get_user_by_token(db, token)
- if not user:
- raise PermissionDeniedError("Invalid or expired token")
- return user
|