authentication.py

from fastapi import HTTPException, status
from passlib.context import CryptContext
import jwt
# from dotenv import dotenv_values
from models import User
import os

credentials = {
    "EMAIL": os.getenv("EMAIL"),
    "PASSWORD": os.getenv("PASSWORD"),
    "SECRET": os.getenv("SECRET"),
    "SERVER_URL": os.getenv("SERVER_URL"),
}

# credentials = dotenv_values(".env")

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


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


async def verify_token(token: str):
    try:
        payload = jwt.decode(token, credentials["SECRET"], algorithms=["HS256"])
        user = await User.get(id=payload["id"])
    except ValueError:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid token",
            headers={"WWW-Authenticate": "Bearer"}
        )
    return user


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


async def authenticate_user(username: str, password: str):
    user = await User.get(username=username)
    if user and await verify_password(password, user.password):
        return user
    return False


async def token_generator(username: str, password: str):
    user = await authenticate_user(username=username, password=password)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="Invalid username or password",
            headers={"WWW-Authenticate": "Bearer"}
        )
    token_data = {"id": user.id, "username": user.username}
    token = jwt.encode(token_data, credentials["SECRET"], algorithm="HS256")
    return token