generated file: dependencies/auth.py

Author: coslynx

Diff for: dependencies/auth.py

@@ -0,0 +1,83 @@
+from fastapi import Depends, HTTPException, status
+from fastapi.security import OAuth2PasswordBearer
+from jose import JWTError, jwt
+from sqlalchemy.orm import Session
+
+from . import schemas
+from .config import settings
+from .database import get_db
+from .models import User
+
+oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
+
+SECRET_KEY = settings.secret_key
+ALGORITHM = settings.algorithm
+ACCESS_TOKEN_EXPIRE_MINUTES = settings.access_token_expire_minutes
+
+
+def create_access_token(data: dict) -> str:
+    """
+    Generates a JWT access token.
+
+    Args:
+        data (dict): A dictionary containing user data to include in the token.
+
+    Returns:
+        str: The encoded JWT access token.
+    """
+    to_encode = data.copy()
+    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def verify_access_token(token: str, credentials_exception) -> schemas.TokenData:
+    """
+    Verifies a JWT access token.
+
+    Args:
+        token (str): The JWT access token to verify.
+        credentials_exception: An HTTPException to raise if the token is invalid.
+
+    Returns:
+        schemas.TokenData: The decoded token data.
+
+    Raises:
+        HTTPException: If the token is invalid or expired.
+    """
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        id: str = payload.get("sub")
+        if id is None:
+            raise credentials_exception
+        token_data = schemas.TokenData(id=id)
+    except JWTError:
+        raise credentials_exception
+    return token_data
+
+
+def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
+    """
+    Retrieves the current user from the database using the provided access token.
+
+    Args:
+        token (str): The JWT access token.
+        db (Session): The database session.
+
+    Returns:
+        User: The current user object.
+
+    Raises:
+        HTTPException: If the token is invalid or the user is not found.
+    """
+    credentials_exception = HTTPException(
+        status_code=status.HTTP_401_UNAUTHORIZED,
+        detail="Could not validate credentials",
+        headers={"WWW-Authenticate": "Bearer"},
+    )
+    token_data = verify_access_token(token, credentials_exception)
+    user = db.query(User).filter(User.id == token_data.id).first()
+    if user is None:
+        raise credentials_exception
+    return user