generated file: utils/auth_utils.py

coslynx · coslynx · commit 61a26f8f700f · 2024-10-18T19:17:53.000+05:30
diff --git a/utils/auth_utils.py b/utils/auth_utils.py
@@ -0,0 +1,54 @@
+from datetime import datetime, timedelta
+import jwt
+from fastapi import HTTPException, status, Depends
+from fastapi.security import OAuth2PasswordBearer
+from sqlalchemy.orm import Session
+
+from database import get_db
+from . import models, schemas
+from dotenv import load_dotenv
+
+load_dotenv()
+SECRET_KEY = os.getenv("JWT_SECRET_KEY")
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+def create_access_token(data: dict, expires_delta: timedelta = timedelta(minutes=15)) -> str:
+    """Generates a JWT access token for a user.
+
+    Args:
+        data (dict): A dictionary containing user information to be encoded in the token.
+        expires_delta (timedelta, optional): The expiration time for the token. Defaults to timedelta(minutes=15).
+
+    Returns:
+        str: The encoded JWT access token.
+    """
+    to_encode = data.copy()
+    expire = datetime.utcnow() + expires_delta
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm="HS256")
+    return encoded_jwt
+
+async def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> models.User:
+    """Verifies the JWT token and retrieves the associated user from the database.
+
+    Args:
+        token (str): The JWT token provided in the Authorization header.
+        db (Session): The SQLAlchemy database session.
+
+    Returns:
+        models.User: The user object associated with the token.
+
+    Raises:
+        HTTPException: If the token is invalid, expired, or the user is not found.
+    """
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=["HS256"])
+        user = db.query(models.User).filter(models.User.email == payload["sub"]).first()
+        if not user:
+            raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
+        return user
+    except jwt.ExpiredSignatureError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token expired")
+    except jwt.InvalidTokenError:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
