(self, email: str, password: str)
| 22 | self.app_email_service = app_email_service |
| 23 | |
| 24 | async def login(self, email: str, password: str) -> TokenBearer: |
| 25 | user = await self.user_service.one( |
| 26 | filters=[ |
| 27 | Filter("email", Oper.EQ, email), |
| 28 | ] |
| 29 | ) |
| 30 | if user is None: |
| 31 | raise UnauthorizedException(error_no=ErrorNo.USER_EMAIL_NOT_FOUND, message="Unauthorized!") |
| 32 | if user.status != UserStatus.ACTIVE: |
| 33 | raise UnauthorizedException(error_no=ErrorNo.USER_STATUS_NOT_ACTIVE, message="Unauthorized!") |
| 34 | |
| 35 | if not self.hash_service.verify_password(password=password, hashed_password=user.hash_password): |
| 36 | raise UnauthorizedException(error_no=ErrorNo.AUTHORIZATION_USER_PASSWORD_INVALID, message="Unauthorized!") |
| 37 | user = await self.user_service.update(uid=user.id, data={"session": self.hash_service.random_string()}) |
| 38 | |
| 39 | return self.hash_service.create_token_bearer(user=user) |
| 40 | |
| 41 | async def signup( |
| 42 | self, |
no test coverage detected