Coverage for app/api/deps.py: 89%

1from collections.abc import Generator 

2from typing import Annotated 

3 

4import jwt 

5from fastapi import Depends, HTTPException, status 

6from fastapi.security import OAuth2PasswordBearer 

7from jwt.exceptions import InvalidTokenError 

8from pydantic import ValidationError 

9from sqlmodel import Session 

10 

11from app.core import security 

12from app.core.config import settings 

13from app.core.db import engine 

14from app.models import TokenPayload, User 

15 

16reusable_oauth2 = OAuth2PasswordBearer( 

17 tokenUrl=f"{settings.API_V1_STR}/login/access-token" 

18) 

19 

20 

21def get_db() -> Generator[Session, None, None]: 

22 with Session(engine) as session: 

23 yield session 

24 

25 

26SessionDep = Annotated[Session, Depends(get_db)] 

27TokenDep = Annotated[str, Depends(reusable_oauth2)] 

28 

29 

30def get_current_user(session: SessionDep, token: TokenDep) -> User: 

31 try: 

32 payload = jwt.decode( 

33 token, settings.SECRET_KEY, algorithms=[security.ALGORITHM] 

34 ) 

35 token_data = TokenPayload(**payload) 

36 except (InvalidTokenError, ValidationError): 

37 raise HTTPException( 

38 status_code=status.HTTP_403_FORBIDDEN, 

39 detail="Could not validate credentials", 

40 ) 

41 user = session.get(User, token_data.sub) 

42 if not user: 

43 raise HTTPException(status_code=404, detail="User not found") 

44 if not user.is_active: 

45 raise HTTPException(status_code=400, detail="Inactive user") 

46 return user 

47 

48 

49CurrentUser = Annotated[User, Depends(get_current_user)] 

50 

51 

52def get_current_active_superuser(current_user: CurrentUser) -> User: 

53 if not current_user.is_superuser: 

54 raise HTTPException( 

55 status_code=403, detail="The user doesn't have enough privileges" 

56 ) 

57 return current_user