| 1234567891011121314151617181920212223242526 |
- from fastapi import Depends, HTTPException, status
- from fastapi.security import OAuth2PasswordBearer
- import auth_utils
- async def get_current_user(token: str = Depends(auth_utils.oauth2_scheme)):
- payload = auth_utils.decode_token(token)
- if not payload:
- raise HTTPException(
- status_code=status.HTTP_401_UNAUTHORIZED,
- detail="Could not validate credentials",
- headers={"WWW-Authenticate": "Bearer"},
- )
- return payload
- async def require_admin(current_user: dict = Depends(get_current_user)):
- if current_user.get("role") != 'admin':
- raise HTTPException(
- status_code=status.HTTP_403_FORBIDDEN,
- detail="Admin role required"
- )
- return current_user
- async def get_current_user_optional(token: str = Depends(auth_utils.oauth2_scheme_optional)):
- if not token:
- return None
- return auth_utils.decode_token(token)
|