Coverage for opt/mealie/lib/python3.12/site-packages/mealie/routes/_base/base_controllers.py: 82%

114 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 17:29 +0000

1from abc import ABC 1a

2from logging import Logger 1a

3 

4from fastapi import Depends, HTTPException 1a

5from pydantic import UUID4, ConfigDict 1a

6from sqlalchemy.orm import Session 1a

7 

8from mealie.core.config import get_app_dirs, get_app_settings 1a

9from mealie.core.dependencies.dependencies import ( 1a

10 get_admin_user, 

11 get_current_user, 

12 get_integration_id, 

13 get_public_group, 

14) 

15from mealie.core.exceptions import mealie_registered_exceptions 1a

16from mealie.core.root_logger import get_logger 1a

17from mealie.core.settings.directories import AppDirectories 1a

18from mealie.core.settings.settings import AppSettings 1a

19from mealie.db.db_setup import generate_session 1a

20from mealie.lang import local_provider 1a

21from mealie.lang.providers import Translator 1a

22from mealie.repos._utils import NOT_SET, NotSet 1a

23from mealie.repos.all_repositories import AllRepositories, get_repositories 1a

24from mealie.routes._base.checks import OperationChecks 1a

25from mealie.schema.household.household import HouseholdInDB 1a

26from mealie.schema.user.user import GroupInDB, PrivateUser 1a

27from mealie.services.event_bus_service.event_bus_service import EventBusService 1a

28from mealie.services.event_bus_service.event_types import EventDocumentDataBase, EventTypes 1a

29 

30 

31class _BaseController(ABC): # noqa: B024 1a

32 session: Session = Depends(generate_session) 1a

33 translator: Translator = Depends(local_provider) 1a

34 

35 _repos: AllRepositories | None = None 1a

36 _logger: Logger | None = None 1a

37 _settings: AppSettings | None = None 1a

38 _folders: AppDirectories | None = None 1a

39 

40 @property 1a

41 def t(self): 1a

42 return self.translator.t if self.translator else local_provider().t 1CO678ycPQdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKB9b

43 

44 @property 1a

45 def repos(self): 1a

46 if not self._repos: 1)!*+,RCO6780yc#$1S-.2%'(TU/PQ3:VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKB;9b

47 self._repos = AllRepositories(self.session, group_id=self.group_id, household_id=self.household_id) 1)!*+,RCO6780yc#$1S-.2%'(TU/PQ3:VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKB;9b

48 return self._repos 1)!*+,RCO6780yc#$1S-.2%'(TU/PQ3:VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKB;9b

49 

50 @property 1a

51 def logger(self) -> Logger: 1a

52 if not self._logger: 1=RC0ycSTUPQ3VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKBb

53 self._logger = get_logger() 1=RC0ycSTUPQ3VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKBb

54 return self._logger 1=RC0ycSTUPQ3VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKBb

55 

56 @property 1a

57 def settings(self) -> AppSettings: 1a

58 if not self._settings: 58 ↛ 60line 58 didn't jump to line 60 because the condition on line 58 was always true1Oyc?b

59 self._settings = get_app_settings() 1Oyc?b

60 return self._settings 1Oyc?b

61 

62 @property 1a

63 def folders(self) -> AppDirectories: 1a

64 if not self._folders: 64 ↛ 66line 64 didn't jump to line 66 because the condition on line 64 was always true

65 self._folders = get_app_dirs() 

66 return self._folders 

67 

68 @property 1a

69 def group_id(self) -> UUID4 | None | NotSet: 1a

70 return NOT_SET 

71 

72 @property 1a

73 def household_id(self) -> UUID4 | None | NotSet: 1a

74 return NOT_SET 

75 

76 model_config = ConfigDict(arbitrary_types_allowed=True) 1a

77 

78 

79class BasePublicController(_BaseController): 1a

80 """ 

81 This is a public class for all User restricted controllers in the API. 

82 It includes the common SharedDependencies and some common methods used 

83 by all Admin controllers. 

84 """ 

85 

86 ... 1a

87 

88 

89class BasePublicGroupExploreController(BasePublicController): 1a

90 """ 

91 Base class for all controllers that are public and explore group data. 

92 """ 

93 

94 group: GroupInDB = Depends(get_public_group) 1a

95 

96 @property 1a

97 def group_id(self) -> UUID4 | None | NotSet: 1a

98 return self.group.id 

99 

100 def get_public_household(self, household_slug_or_id: str | UUID4) -> HouseholdInDB: 1a

101 household = self.repos.households.get_by_slug_or_id(household_slug_or_id) 

102 if not household or household.preferences.private_household: 

103 raise HTTPException(404, "household not found") 

104 return household 

105 

106 def get_explore_url_path(self, endpoint: str) -> str: 1a

107 if endpoint.startswith("/"): 

108 endpoint = endpoint[1:] 

109 return f"/explore/groups/{self.group.slug}/{endpoint}" 

110 

111 

112class BasePublicHouseholdExploreController(BasePublicGroupExploreController): 1a

113 """ 

114 Base class for all controllers that are public and explore household data. 

115 """ 

116 

117 @property 1a

118 def cross_household_repos(self): 1a

119 """ 

120 Household-level repos with no household filter. Public controllers don't have access to a household identifier; 

121 instead, they return all public data, filtered by the household preferences. 

122 

123 When using this repo, the caller should filter by household preferences, e.g.: 

124 

125 `household.preferences.privateHousehold = FALSE` 

126 """ 

127 return get_repositories(self.session, group_id=self.group_id, household_id=None) 

128 

129 

130class BaseUserController(_BaseController): 1a

131 """ 

132 This is a base class for all User restricted controllers in the API. 

133 It includes the common SharedDependencies and some common methods used 

134 by all Admin controllers. 

135 """ 

136 

137 user: PrivateUser = Depends(get_current_user) 1a

138 integration_id: str = Depends(get_integration_id) 1a

139 translator: Translator = Depends(local_provider) 1a

140 

141 # Manual Cache 

142 _checks: OperationChecks 1a

143 

144 def registered_exceptions(self, ex: type[Exception]) -> str: 1a

145 registered = { 1CdzefghiAjklmnJopqrstuGIvwxBb

146 **mealie_registered_exceptions(self.translator), 

147 } 

148 return registered.get(ex, self.t("generic.server-error")) 1CdzefghiAjklmnJopqrstuGIvwxBb

149 

150 @property 1a

151 def group_id(self) -> UUID4: 1a

152 return self.user.group_id 1)!*+,RCO6780y@[c#$1S-.2%'(TU/PQ]^3:VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKB;9b

153 

154 @property 1a

155 def household_id(self) -> UUID4: 1a

156 return self.user.household_id 1)!*+,RCO6780yc#$1S-.2%'(TU/PQ3:VWXY45ZdzeDfghiAjEklFmnJopqrstuLMGNHIvwxKB;9b

157 

158 @property 1a

159 def group(self) -> GroupInDB: 1a

160 return self.repos.groups.get_one(self.group_id) 1!c#$1deDfghiAjEklFmnopqrstuHvwxb

161 

162 @property 1a

163 def household(self) -> HouseholdInDB: 1a

164 return self.repos.households.get_one(self.household_id) 1Rc1S%'(TUVWXZdzeDfghiAjEklFmnopqrstuHvwxKBb

165 

166 @property 1a

167 def checks(self) -> OperationChecks: 1a

168 if not self._checks: 168 ↛ 170line 168 didn't jump to line 170 because the condition on line 168 was always true12b

169 self._checks = OperationChecks(self.user) 12b

170 return self._checks 12b

171 

172 

173class BaseAdminController(BaseUserController): 1a

174 """ 

175 This is a base class for all Admin restricted controllers in the API. 

176 It includes the common Shared Dependencies and some common methods used 

177 by all Admin controllers. 

178 """ 

179 

180 user: PrivateUser = Depends(get_admin_user) 1a

181 

182 @property 1a

183 def repos(self): 1a

184 if not self._repos: 

185 # Admins have access to all groups and households, so we don't want to filter by group_id or household_id 

186 self._repos = AllRepositories(self.session, group_id=None, household_id=None) 

187 return self._repos 

188 

189 

190class BaseCrudController(BaseUserController): 1a

191 """ 

192 Base class for all CRUD controllers to facilitate common CRUD functions. 

193 """ 

194 

195 event_bus: EventBusService = Depends(EventBusService.as_dependency) 1a

196 

197 def publish_event( 1a

198 self, 

199 event_type: EventTypes, 

200 document_data: EventDocumentDataBase, 

201 group_id: UUID4, 

202 household_id: UUID4 | None, 

203 message: str = "", 

204 ) -> None: 

205 self.event_bus.dispatch( 1C678yc1YdzeDfghijEklFmnopqrstuLMGNHIvwxKB9b

206 integration_id=self.integration_id, 

207 group_id=group_id, 

208 household_id=household_id, 

209 event_type=event_type, 

210 document_data=document_data, 

211 message=message, 

212 )