from datetime import datetime, timedelta from typing import Any, Union from jose import jwt from passlib.context import CryptContext from app.core.config import settings # Password hashing context pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") # Token types TOKEN_TYPE_ACCESS = "access" TOKEN_TYPE_REFRESH = "refresh" def create_access_token( subject: Union[str, Any], expires_delta: timedelta = None ) -> str: """ Create a JWT access token for a given subject. Args: subject: Subject of the token (typically user ID) expires_delta: Optional expiration time delta Returns: Encoded JWT token """ if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta( minutes=settings.ACCESS_TOKEN_EXPIRE_MINUTES ) to_encode = { "exp": expire, "sub": str(subject), "type": TOKEN_TYPE_ACCESS } encoded_jwt = jwt.encode( to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM ) return encoded_jwt def create_refresh_token( subject: Union[str, Any], expires_delta: timedelta = None ) -> str: """ Create a JWT refresh token for a given subject. Args: subject: Subject of the token (typically user ID) expires_delta: Optional expiration time delta Returns: Encoded JWT token """ if expires_delta: expire = datetime.utcnow() + expires_delta else: expire = datetime.utcnow() + timedelta( days=settings.REFRESH_TOKEN_EXPIRE_DAYS ) to_encode = { "exp": expire, "sub": str(subject), "type": TOKEN_TYPE_REFRESH } encoded_jwt = jwt.encode( to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM ) return encoded_jwt def verify_password(plain_password: str, hashed_password: str) -> bool: """ Verify a plain password against a hashed password. Args: plain_password: Plain password hashed_password: Hashed password Returns: True if passwords match, False otherwise """ return pwd_context.verify(plain_password, hashed_password) def get_password_hash(password: str) -> str: """ Hash a password. Args: password: Plain password Returns: Hashed password """ return pwd_context.hash(password)