from datetime import datetime, timedelta from typing import Optional from passlib.context import CryptContext from jose import JWTError, jwt from app.config import settings from app.models import User from sqlalchemy.orm import Session pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") def verify_password(plain_password: str, hashed_password: str) -> bool: return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: return pwd_context.hash(password) def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str: 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.jwt_secret_key, algorithm=settings.jwt_algorithm ) return encoded_jwt def verify_token(token: str) -> Optional[int]: try: payload = jwt.decode( token, settings.jwt_secret_key, algorithms=[settings.jwt_algorithm] ) user_id: int = payload.get("sub") if user_id is None: return None return user_id except JWTError: return None def authenticate_user(db: Session, email: str, password: str) -> Optional[User]: user = db.query(User).filter(User.email == email).first() if not user or not verify_password(password, user.hashed_password): return None return user