Source code for geochemistrypi.auth.utils

from datetime import datetime, timedelta
from typing import Union

from fastapi import HTTPException, status
from jose import JWTError, jwt
from passlib.context import CryptContext

from .constants import ALGORITHM, SECRET_KEY

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")




def get_password_hash(password):
    return pwd_context.hash(password)





def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)





def create_access_token(data: dict, expires_delta: Union[timedelta, None] = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt





def decode_access_token(token: str):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        # username: str = payload.get("sub")
        email: str = payload.get("sub")
        # if username is None:
        if email is None:
            raise credentials_exception
        # token_data = TokenData(username=username)
        # token_data = TokenData(email=email)
        token_data = email
    except JWTError:
        raise credentials_exception
    return token_data