108 lines
4.4 KiB
Python
108 lines
4.4 KiB
Python
from typing import Optional, Tuple, List
|
|
|
|
from backend.services.permission.permission_service import PermissionService
|
|
from backend.services.permission.role_service import RoleService
|
|
from common.constants.region import UserRegion
|
|
from common.log.log_utils import log_entry_exit_async
|
|
from backend.business.signin_manager import SignInManager
|
|
from backend.models.user.constants import UserLoginAction
|
|
|
|
|
|
class SignInHub:
|
|
def __init__(self) -> None:
|
|
self.signin_manager = SignInManager()
|
|
# TODO: Dax - Event dispatch and notification center
|
|
# self.notification_center = NotificationCenter(sender_id=settings.SYSTEM_USER_ID)
|
|
# self.event_dispatcher = UserEventDispatcher(owner_id=settings.SYSTEM_USER_ID)
|
|
|
|
@log_entry_exit_async
|
|
async def signin_with_email_and_code(
|
|
self, email: str, code: str, host: str, time_zone: Optional[str] = "UTC"
|
|
) -> Tuple[UserLoginAction, Optional[int], Optional[str], Optional[str], Optional[UserRegion], Optional[List[str]],
|
|
Optional[List[str]]]:
|
|
"""
|
|
Interacts with the business layer to handle the sign-in process with email and code.
|
|
Try to signin with email and code.
|
|
create a new user account, if the email address has never been used before.
|
|
|
|
Args:
|
|
email (str): email address
|
|
code (str): auth code to be verified
|
|
host (str): the host address by which the client access the frontend service
|
|
time_zone (Optional[str]): time zone of the frontend service
|
|
Returns:
|
|
[int, Optional[int], Optional[str], Optional[str]]:
|
|
- int: UserLoginAction
|
|
- Optional[int]: user role
|
|
- Optional[str]: user_id
|
|
- Optional[str]: flid
|
|
- Optional[str]: region
|
|
- Optional[str]: user role names
|
|
- Optional[str]: user permission keys
|
|
"""
|
|
return await self.signin_manager.signin_with_email_and_code(
|
|
email=email, code=code, host=host, time_zone=time_zone
|
|
)
|
|
|
|
@log_entry_exit_async
|
|
async def signin_with_email_and_password(
|
|
self, email: str, password: str
|
|
) -> Tuple[UserLoginAction, Optional[int], Optional[str], Optional[str], Optional[List[str]], Optional[List[str]]]:
|
|
"""Try to signin with email and password.
|
|
|
|
Args:
|
|
email (str): email address
|
|
password (str): password to be verified
|
|
|
|
Returns:
|
|
[int, Optional[int], Optional[str], Optional[str]]:
|
|
- int: UserLoginAction
|
|
- Optional[int]: user role
|
|
- Optional[str]: user_id
|
|
- Optional[str]: flid
|
|
- Optional[List[str]]: user role names
|
|
- Optional[List[str]]: user permission keys
|
|
"""
|
|
return await self.signin_manager.signin_with_email_and_password(
|
|
email=email, password=password
|
|
)
|
|
|
|
@log_entry_exit_async
|
|
async def update_new_user_flid(
|
|
self, user_id: str, user_flid: str
|
|
) -> Tuple[UserLoginAction, Optional[str]]:
|
|
return await self.signin_manager.update_new_user_flid(
|
|
user_id=user_id, user_flid=user_flid
|
|
)
|
|
|
|
@log_entry_exit_async
|
|
async def try_signin_with_email(self, email: str, host: str) -> UserLoginAction:
|
|
return await self.signin_manager.try_signin_with_email(email=email, host=host)
|
|
|
|
@log_entry_exit_async
|
|
async def reset_password_through_email(self, email: str, host: str) -> int:
|
|
return await self.signin_manager.reset_password_through_email(
|
|
email=email, host=host
|
|
)
|
|
|
|
@log_entry_exit_async
|
|
async def update_user_password(self, user_id: str, password: str) -> dict[str, any]:
|
|
return await self.signin_manager.update_user_password(
|
|
user_id=user_id, password=password
|
|
)
|
|
|
|
@log_entry_exit_async
|
|
async def send_email_code(self, sender_id: str, email: str) -> dict[str, any]:
|
|
result = await self.signin_manager.send_email_code(sender_id, email)
|
|
return {"succeeded": result}
|
|
|
|
@log_entry_exit_async
|
|
async def send_mobile_code(self, sender_id: str, mobile: str) -> dict[str, any]:
|
|
result = await self.signin_manager.send_mobile_code(sender_id, mobile)
|
|
return {"succeeded": result}
|
|
|
|
@log_entry_exit_async
|
|
async def sign_out(self, identity: str) -> bool:
|
|
# TODO: to be implemented
|
|
return True
|