Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/user_services/registration_service.py: 25%

1from logging import Logger 

2from uuid import uuid4 

3 

4from fastapi import HTTPException, status 

5from pydantic import UUID4 

6 

7from mealie.core.config import get_app_settings 

8from mealie.core.security import hash_password 

9from mealie.lang.providers import Translator 

10from mealie.repos.all_repositories import get_repositories 

11from mealie.repos.repository_factory import AllRepositories 

12from mealie.schema.group.group_preferences import CreateGroupPreferences 

13from mealie.schema.household.household import HouseholdCreate, HouseholdInDB 

14from mealie.schema.household.household_preferences import CreateHouseholdPreferences 

15from mealie.schema.user.registration import CreateUserRegistration 

16from mealie.schema.user.user import GroupBase, GroupInDB, PrivateUser, UserIn 

17from mealie.services.group_services.group_service import GroupService 

18from mealie.services.household_services.household_service import HouseholdService 

19from mealie.services.seeder.seeder_service import SeederService 

20 

21 

22class RegistrationService: 

23 logger: Logger 

24 repos: AllRepositories 

25 

26 def __init__(self, logger: Logger, db: AllRepositories, translator: Translator): 

27 self.logger = logger 

28 self.repos = db 

29 self.t = translator.t 

30 

31 def _create_new_user(self, group: GroupInDB, household: HouseholdInDB, new_group: bool) -> PrivateUser: 

32 new_user = UserIn( 

33 email=self.registration.email, 

34 username=self.registration.username, 

35 password=hash_password(self.registration.password), 

36 full_name=self.registration.full_name, 

37 advanced=self.registration.advanced, 

38 group=group, 

39 household=household, 

40 can_invite=new_group, 

41 can_manage=new_group, 

42 can_manage_household=new_group, 

43 can_organize=new_group, 

44 ) 

45 

46 # TODO: problem with repository type, not type here 

47 return self.repos.users.create(new_user) # type: ignore 

48 

49 def _register_new_group(self) -> GroupInDB: 

50 group_data = GroupBase(name=self.registration.group) 

51 

52 group_preferences = CreateGroupPreferences( 

53 group_id=uuid4(), 

54 private_group=self.registration.private, 

55 ) 

56 

57 return GroupService.create_group(self.repos, group_data, group_preferences) 

58 

59 def _fetch_or_register_new_household(self, group_id: UUID4) -> HouseholdInDB: 

60 settings = get_app_settings() 

61 new_household_name = self.registration.household or settings.DEFAULT_HOUSEHOLD 

62 

63 group_repos = get_repositories(self.repos.session, group_id=group_id) 

64 household_fetch = group_repos.households.get_by_name(new_household_name) 

65 if household_fetch: 

66 return household_fetch 

67 

68 household_data = HouseholdCreate(name=new_household_name) 

69 household_preferences = CreateHouseholdPreferences( 

70 private_household=self.registration.private, 

71 first_day_of_week=0, 

72 recipe_public=not self.registration.private, 

73 recipe_show_nutrition=self.registration.advanced, 

74 recipe_show_assets=self.registration.advanced, 

75 recipe_landscape_view=False, 

76 recipe_disable_comments=self.registration.advanced, 

77 ) 

78 return HouseholdService.create_household(group_repos, household_data, household_preferences) 

79 

80 def register_user(self, registration: CreateUserRegistration) -> PrivateUser: 

81 self.registration = registration 

82 

83 if self.repos.users.get_by_username(registration.username): 

84 raise HTTPException(status.HTTP_409_CONFLICT, {"message": self.t("exceptions.username-conflict-error")}) 

85 elif self.repos.users.get_one(registration.email, "email"): 

86 raise HTTPException(status.HTTP_409_CONFLICT, {"message": self.t("exceptions.email-conflict-error")}) 

87 

88 token_entry = None 

89 new_group = False 

90 

91 if registration.group_token: 

92 token_entry = self.repos.group_invite_tokens.get_one(registration.group_token) 

93 if not token_entry: 

94 raise HTTPException(status.HTTP_400_BAD_REQUEST, {"message": "Invalid group token"}) 

95 

96 maybe_none_group = self.repos.groups.get_one(token_entry.group_id) 

97 if maybe_none_group is None: 

98 raise HTTPException(status.HTTP_400_BAD_REQUEST, {"message": "Invalid group token"}) 

99 group = maybe_none_group 

100 

101 maybe_none_household = self.repos.households.get_one(token_entry.household_id) 

102 if maybe_none_household is None: 

103 raise HTTPException(status.HTTP_400_BAD_REQUEST, {"message": "Invalid group token"}) 

104 household = maybe_none_household 

105 elif registration.group: 

106 new_group = True 

107 group = self._register_new_group() 

108 household = self._fetch_or_register_new_household(group.id) 

109 else: 

110 raise HTTPException(status.HTTP_400_BAD_REQUEST, {"message": "Missing group"}) 

111 

112 self.logger.info(f"Registering user {registration.username}") 

113 user = self._create_new_user(group, household, new_group) 

114 

115 if new_group and registration.seed_data: 

116 seeder_service = SeederService(self.repos) 

117 seeder_service.seed_foods(registration.locale) 

118 seeder_service.seed_labels(registration.locale) 

119 seeder_service.seed_units(registration.locale) 

120 

121 if token_entry and user: 

122 token_entry.uses_left = token_entry.uses_left - 1 

123 

124 if token_entry.uses_left == 0: 

125 self.repos.group_invite_tokens.delete(token_entry.token) 

126 

127 else: 

128 self.repos.group_invite_tokens.update(token_entry.token, token_entry) 

129 

130 return user