Coverage for app / routers / auth_router.py: 100%

1from typing import Annotated 

2 

3from fastapi import APIRouter, Depends, HTTPException, status 

4from fastapi.security import OAuth2PasswordRequestForm 

5from sqlalchemy.exc import SQLAlchemyError 

6from sqlalchemy.ext.asyncio import AsyncSession 

7 

8from app.core.database import get_db 

9from app.core.exceptions import AlreadyExistsError, AuthentificationError 

10from app.schemas.user_schemas import AccessTokenInfo, UserRegister, UserRegisterResponse 

11from app.services.user_service import login_user_service, register_user_service 

12 

13router = APIRouter( 

14 prefix="/auth", 

15 tags=["auth"], 

16) 

17 

18 

19@router.post("/register") 

20async def register( 

21 new_user: UserRegister, 

22 db: Annotated[AsyncSession, Depends(get_db)], 

23) -> UserRegisterResponse: 

24 """ 

25 Вызывает сервис для регистрации пользователя. 

26 

27 Возвращает идентификатор пользователя и адрес электронной почты. 

28 """ 

29 try: 

30 user = await register_user_service( 

31 db=db, email=new_user.email, password=new_user.password 

32 ) 

33 return user 

34 

35 except AlreadyExistsError as e: 

36 raise HTTPException(status_code=status.HTTP_409_CONFLICT, detail=str(e)) 

37 

38 except SQLAlchemyError: 

39 raise HTTPException( 

40 status.HTTP_500_INTERNAL_SERVER_ERROR, 

41 detail="Database error", 

42 ) 

43 

44 

45@router.post("/login") 

46async def login( 

47 form_data: Annotated[OAuth2PasswordRequestForm, Depends()], 

48 db: Annotated[AsyncSession, Depends(get_db)], 

49) -> AccessTokenInfo: 

50 """ 

51 Вызывает сервис для проверки электронной почты и пароля пользователя. 

52 

53 Возвращает созданный JWT-токен и тип токена "Bearer". 

54 """ 

55 try: 

56 access_token = await login_user_service( 

57 db=db, 

58 email=form_data.username, 

59 password=form_data.password, 

60 ) 

61 except AuthentificationError as e: 

62 raise HTTPException( 

63 status_code=status.HTTP_401_UNAUTHORIZED, 

64 detail=str(e), 

65 ) 

66 

67 return AccessTokenInfo(access_token=access_token, token_type="Bearer")