"""Dependency injection utilities.""" from typing import Annotated, Generator from uuid import UUID from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from sqlalchemy.orm import Session from app.auth.jwt import decode_access_token from app.database.models.user import User from app.database.session import get_db # Database session dependency DatabaseSession = Annotated[Session, Depends(get_db)] # Security scheme for JWT Bearer token security = HTTPBearer() def get_current_user( credentials: HTTPAuthorizationCredentials = Depends(security), db: Session = Depends(get_db) ) -> User: """ Get current authenticated user from JWT token. Args: credentials: HTTP Authorization Bearer token db: Database session Returns: Current authenticated user Raises: HTTPException: If token is invalid or user not found """ # Decode token token = credentials.credentials payload = decode_access_token(token) if payload is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid authentication credentials", headers={"WWW-Authenticate": "Bearer"}, ) # Extract user ID from token user_id_str: str = payload.get("sub") if user_id_str is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token payload", headers={"WWW-Authenticate": "Bearer"}, ) try: user_id = UUID(user_id_str) except ValueError: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid user ID in token", headers={"WWW-Authenticate": "Bearer"}, ) # Get user from database user = db.query(User).filter(User.id == user_id).first() if user is None: raise HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found", headers={"WWW-Authenticate": "Bearer"}, ) if not user.is_active: raise HTTPException( status_code=status.HTTP_403_FORBIDDEN, detail="User account is deactivated" ) return user