Spaces:
Sleeping
Sleeping
from fastapi import Depends, HTTPException, status | |
from fastapi.security import OAuth2PasswordBearer | |
from jose import jwt, JWTError | |
from .utils.db import tinydb_helper # Ensure correct import path | |
from .utils.jwt_utils import SECRET_KEY, ALGORITHM # Ensure these are defined in our jwt_utils.py | |
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token") | |
def decode_access_token(token: str, credentials_exception) -> dict: | |
try: | |
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) | |
user_id: str = payload.get("sub") | |
name: str = payload.get("name") | |
role: str = payload.get("role") | |
if user_id is None or name is None or role is None: | |
raise credentials_exception | |
return {"user_id": user_id, "name": name, "role": role} | |
except jwt.PyJWTError: | |
raise credentials_exception | |
async def get_current_user(token: str = Depends(oauth2_scheme)) -> dict: | |
credentials_exception = HTTPException( | |
status_code=status.HTTP_401_UNAUTHORIZED, | |
detail="Could not validate credentials", | |
headers={"WWW-Authenticate": "Bearer"}, | |
) | |
return decode_access_token(token, credentials_exception) | |