Class AuthService

src/app/auth/service/auth_service.py:13–122

class AuthService:
    def __init__(
        self,
        user_service: UserService,
        hash_service: HashService,
        app_email_service: AppMailService,
    ) -> None:
        self.user_service = user_service
        self.hash_service = hash_service
        self.app_email_service = app_email_service

    async def login(self, email: str, password: str) -> TokenBearer:
        user = await self.user_service.one(
            filters=[
                Filter("email", Oper.EQ, email),
            ]
        )
        if user is None:
            raise UnauthorizedException(error_no=ErrorNo.USER_EMAIL_NOT_FOUND, message="Unauthorized!")
        if user.status != UserStatus.ACTIVE:
            raise UnauthorizedException(error_no=ErrorNo.USER_STATUS_NOT_ACTIVE, message="Unauthorized!")

        if not self.hash_service.verify_password(password=password, hashed_password=user.hash_password):
            raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_USER_PASSWORD_INVALID, message="Unauthorized!")
        user = await self.user_service.update(uid=user.id, data={"session": self.hash_service.random_string()})

        return self.hash_service.create_token_bearer(user=user)

    async def signup(
        self,
        first_name: str,
        second_name: str,
        email: str,
        password: str,
    ) -> User:
        user = await self.user_service.one(
            filters=[
                Filter("email", Oper.EQ, email),
            ]
        )
        if user is not None:
            raise UnprocessableEntityException(
                error_no=ErrorNo.USER_EMAIL_ALREADY_EXISTS, message="User already exists"
            )

        user = User(
            first_name=first_name,
            second_name=second_name,
            email=email,
            hash_password=self.hash_service.hash_password(password),
            session=self.hash_service.random_string(),
            status=UserStatus.PENDING,
            roles=[Role.USER],
        )
        user = await self.user_service.create(data=user)
        await self.send_confirm_email(user)
        return user