from datetime import datetime, timedelta from typing import Optional from jose import JWTError, jwt from fastapi import HTTPException, status, Depends from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials from sqlalchemy.orm import Session from ..database.config import settings from ..models.user import User from ..schemas.token import TokenData from ..database.database import get_db security = HTTPBearer() def create_access_token(data: dict, expires_delta: Optional[timedelta] = None): to_encode = data.copy() if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta(minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES) to_encode.update({"exp": expire}) encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM) return encoded_jwt def verify_token(token: str, credentials_exception): try: payload = jwt.decode(token, settings.SECRET_KEY, algorithms=[settings.ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = TokenData(username=username) except JWTError: raise credentials_exception return token_data async def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db) ): credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) token_data = verify_token(credentials.credentials, credentials_exception) user = db.query(User).filter(User.username == token_data.username).first() if user is None: raise credentials_exception return user async def get_current_active_user(current_user: User = Depends(get_current_user)): if not current_user.is_active: raise HTTPException(status_code=400, detail="Inactive user") return current_user