Coverage for app / services / user_service.py: 100%

1from sqlalchemy.ext.asyncio import AsyncSession 

2 

3from app.core.exceptions import AlreadyExistsError, AuthentificationError 

4from app.core.security import create_jwt_token, get_password_hash, verify_password 

5from app.models import User 

6from app.repositories.user_repository import UserRepository 

7 

8 

9async def register_user_service(db: AsyncSession, email: str, password: str) -> User: 

10 """ 

11 Перед регистрацией пользователя проверяет уникальность адреса 

12 электронной почты в базе данных. 

13 

14 - Если пользователь с указанным `email` существует, то вызывается 

15 исключение `AlreadyExistsError`. 

16 """ 

17 user = await UserRepository.get_by_email(db, email) 

18 if user: 

19 raise AlreadyExistsError(f"Email already exists") 

20 new_user = User(email=email, hashed_password=get_password_hash(password)) 

21 try: 

22 db.add(new_user) 

23 await db.commit() 

24 await db.refresh(new_user) 

25 return new_user 

26 except Exception: 

27 await db.rollback() 

28 raise 

29 

30 

31async def login_user_service(db: AsyncSession, email: str, password: str) -> str: 

32 """ 

33 Проверяет корректность адреса электронной почты и пароля и возвращает JWT-токен. 

34 

35 - Если в базе данных нет пользователя с указанным `email` или пароль 

36 неправильный, то вызывается исключение `AuthentificationError`. 

37 """ 

38 user = await UserRepository.get_by_email(db, email) 

39 if not user or not verify_password(password, user.hashed_password): 

40 raise AuthentificationError("Incorrect email or password") 

41 return create_jwt_token(user.id)