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

1from collections.abc import Generator (empty)

2from typing import Annotated (empty)

3 

4import jwt (empty)

5from fastapi import Depends, HTTPException, status (empty)

6from fastapi.security import OAuth2PasswordBearer (empty)

7from jwt.exceptions import InvalidTokenError (empty)

8from pydantic import ValidationError (empty)

9from sqlmodel import Session (empty)

10 

11from app.core import security (empty)

12from app.core.config import settings (empty)

13from app.core.db import engine (empty)

14from app.models import TokenPayload, User (empty)

15 

16reusable_oauth2 = OAuth2PasswordBearer( (empty)

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

18) 

19 

20 

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

22 with Session(engine) as session: 49 ctx1amnopqrstuvwMNOPQRSTxUbcdeyzfghABCDEFGVWiHIJjkKLl

23 yield session 49 ctx1amnopqrstuvwMNOPQRSTxUbcdeyzfghABCDEFGVWiHIJjkKLl

24 

25 

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

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

28 

29 

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

31 try: 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

32 payload = jwt.decode( 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

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

34 ) 

35 token_data = TokenPayload(**payload) 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

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) 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

42 if not user: 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

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

44 if not user.is_active: 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

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

46 return user 37 ctx1mnopqrstuvwxbcdeyzfghABCDEFGiHIJjkKLl

47 

48 

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

50 

51 

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

53 if not current_user.is_superuser: 11 ctx1bcdefghijkl

54 raise HTTPException( 2 ctx1bh

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

56 ) 

57 return current_user 9 ctx1cdefgijkl