from infra.log.module_logger import ModuleLogger from typing import Optional from app.authentication.backend.models.user.constants import ( NewUserMethod, UserAccountProperty, ) from app.authentication.backend.models.user.models import UserAccountDoc from app.authentication.backend.models.permission.constants import ( AdministrativeRole, Capability, ) from app.authentication.backend.infra.auth.user_auth_handler import ( UserAuthHandler, ) from app.authentication.backend.infra.user_profile.user_profile_handler import ( UserProfileHandler, ) from infra.log.log_utils import log_entry_exit_async from infra.models.constants import UserRegion class UserManagementService: def __init__(self) -> None: self.user_auth_handler = UserAuthHandler() self.user_profile_handler = UserProfileHandler() self.module_logger = ModuleLogger(sender_id=UserManagementService) @log_entry_exit_async async def create_new_user_account( self, method: NewUserMethod, region: UserRegion ) -> UserAccountDoc: """create a new user account document in DB Args: method (NewUserMethod): the method the new user came from region : preferred user region detected via the user log-in website Returns: str: id of user account """ if NewUserMethod.EMAIL == method: user_account = await self.user_profile_handler.create_new_user_account( UserAccountProperty.EMAIL_VERIFIED, Capability.VISITOR, AdministrativeRole.PERSONAL, region, ) elif NewUserMethod.MOBILE == method: user_account = await self.user_profile_handler.create_new_user_account( UserAccountProperty.EMAIL_VERIFIED, Capability.VISITOR, AdministrativeRole.PERSONAL, region, ) # Create other doc in collections for the new user # TODO: Should convert to notification # await UserAchievement(str(user_account.id)).create_activeness_achievement() return user_account async def initialize_new_user_data( self, user_id: str, method: NewUserMethod, email_address: str = None, mobile_number: str = None, region: UserRegion = UserRegion.ZH_CN, time_zone: Optional[str] = "UTC", ): """Init data for the new user Args: user_id (str): user id method (NewUserMethod): the method the new user came from Returns: result: True if initilize data for the new user successfully, else return False """ # create basic and provider profile doc for the new user if NewUserMethod.EMAIL == method: await self.user_profile_handler.create_basic_profile( user_id, email_address, True, None, False, False, region, time_zone ) await self.user_auth_handler.save_email_auth_method(user_id, email_address) elif NewUserMethod.MOBILE == method: await self.user_profile_handler.create_basic_profile( user_id, None, False, mobile_number, True, False, region, time_zone ) else: return False await self.user_profile_handler.create_provider_profile(user_id) return True async def get_account_by_id(self, user_id: str) -> UserAccountDoc: return await self.user_profile_handler.get_account_by_id(user_id)