Spaces:
Running
Running
File size: 1,603 Bytes
aa27d2d e000e33 aa27d2d e000e33 aa27d2d e000e33 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | from fastapi import Depends, HTTPException
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from jose import jwt, JWTError
from app.core.config import settings
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.models.user_model import User
from typing import Optional
security = HTTPBearer()
optional_security = HTTPBearer(auto_error=False)
def get_current_user(
credentials: HTTPAuthorizationCredentials = Depends(security),
db: Session = Depends(get_db)
):
token = credentials.credentials
try:
payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
user_id: int = payload.get("user_id")
if user_id is None:
raise HTTPException(status_code=401, detail="Invalid token")
except JWTError:
raise HTTPException(status_code=401, detail="Invalid token")
user = db.query(User).filter(User.id == user_id).first()
if user is None:
raise HTTPException(status_code=401, detail="User not found")
return user
def get_optional_user(
credentials:
Optional[HTTPAuthorizationCredentials] = Depends(optional_security),
db: Session = Depends(get_db)
):
if not credentials:
return None
token = credentials.credentials
try:
payload = jwt.decode(token, settings.JWT_SECRET, algorithms=["HS256"])
user_id = payload.get("user_id")
if not user_id:
return None
return db.query(User).filter(User.id == user_id).first()
except JWTError:
return None |