"""Dependency injection utilities.""" from typing import Annotated 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"}, ) from None # 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