Coverage for opt/mealie/lib/python3.12/site-packages/mealie/services/user_services/registration_service.py: 65%
79 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:32 +0000
1from logging import Logger 1b
2from uuid import uuid4 1b
4from fastapi import HTTPException, status 1b
5from pydantic import UUID4 1b
7from mealie.core.config import get_app_settings 1b
8from mealie.core.security import hash_password 1b
9from mealie.lang.providers import Translator 1b
10from mealie.repos.all_repositories import get_repositories 1b
11from mealie.repos.repository_factory import AllRepositories 1b
12from mealie.schema.group.group_preferences import CreateGroupPreferences 1b
13from mealie.schema.household.household import HouseholdCreate, HouseholdInDB 1b
14from mealie.schema.household.household_preferences import CreateHouseholdPreferences 1b
15from mealie.schema.user.registration import CreateUserRegistration 1b
16from mealie.schema.user.user import GroupBase, GroupInDB, PrivateUser, UserIn 1b
17from mealie.services.group_services.group_service import GroupService 1b
18from mealie.services.household_services.household_service import HouseholdService 1b
19from mealie.services.seeder.seeder_service import SeederService 1b
22class RegistrationService: 1b
23 logger: Logger 1b
24 repos: AllRepositories 1b
26 def __init__(self, logger: Logger, db: AllRepositories, translator: Translator): 1b
27 self.logger = logger 1ac
28 self.repos = db 1ac
29 self.t = translator.t 1ac
31 def _create_new_user(self, group: GroupInDB, household: HouseholdInDB, new_group: bool) -> PrivateUser: 1b
32 new_user = UserIn( 1a
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 )
46 # TODO: problem with repository type, not type here
47 return self.repos.users.create(new_user) # type: ignore 1a
49 def _register_new_group(self) -> GroupInDB: 1b
50 group_data = GroupBase(name=self.registration.group) 1a
52 group_preferences = CreateGroupPreferences( 1a
53 group_id=uuid4(),
54 private_group=self.registration.private,
55 )
57 return GroupService.create_group(self.repos, group_data, group_preferences) 1a
59 def _fetch_or_register_new_household(self, group_id: UUID4) -> HouseholdInDB: 1b
60 settings = get_app_settings() 1a
61 new_household_name = self.registration.household or settings.DEFAULT_HOUSEHOLD 1a
63 group_repos = get_repositories(self.repos.session, group_id=group_id) 1a
64 household_fetch = group_repos.households.get_by_name(new_household_name) 1a
65 if household_fetch: 65 ↛ 68line 65 didn't jump to line 68 because the condition on line 65 was always true1a
66 return household_fetch 1a
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)
80 def register_user(self, registration: CreateUserRegistration) -> PrivateUser: 1b
81 self.registration = registration 1ac
83 if self.repos.users.get_by_username(registration.username): 83 ↛ 84line 83 didn't jump to line 84 because the condition on line 83 was never true1ac
84 raise HTTPException(status.HTTP_409_CONFLICT, {"message": self.t("exceptions.username-conflict-error")})
85 elif self.repos.users.get_one(registration.email, "email"): 85 ↛ 86line 85 didn't jump to line 86 because the condition on line 85 was never true1ac
86 raise HTTPException(status.HTTP_409_CONFLICT, {"message": self.t("exceptions.email-conflict-error")})
88 token_entry = None 1ac
89 new_group = False 1ac
91 if registration.group_token: 1ac
92 token_entry = self.repos.group_invite_tokens.get_one(registration.group_token) 1c
93 if not token_entry: 93 ↛ 96line 93 didn't jump to line 96 because the condition on line 93 was always true1c
94 raise HTTPException(status.HTTP_400_BAD_REQUEST, {"message": "Invalid group token"}) 1c
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
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: 105 ↛ 110line 105 didn't jump to line 110 because the condition on line 105 was always true1a
106 new_group = True 1a
107 group = self._register_new_group() 1a
108 household = self._fetch_or_register_new_household(group.id) 1a
109 else:
110 raise HTTPException(status.HTTP_400_BAD_REQUEST, {"message": "Missing group"})
112 self.logger.info(f"Registering user {registration.username}") 1a
113 user = self._create_new_user(group, household, new_group) 1a
115 if new_group and registration.seed_data: 115 ↛ 116line 115 didn't jump to line 116 because the condition on line 115 was never true1a
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)
121 if token_entry and user: 121 ↛ 122line 121 didn't jump to line 122 because the condition on line 121 was never true1a
122 token_entry.uses_left = token_entry.uses_left - 1
124 if token_entry.uses_left == 0:
125 self.repos.group_invite_tokens.delete(token_entry.token)
127 else:
128 self.repos.group_invite_tokens.update(token_entry.token, token_entry)
130 return user 1a