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

1from abc import ABC 

2from logging import Logger 

3 

4from fastapi import Depends, HTTPException 

5from pydantic import UUID4, ConfigDict 

6from sqlalchemy.orm import Session 

7 

8from mealie.core.config import get_app_dirs, get_app_settings 

9from mealie.core.dependencies.dependencies import ( 

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 

16from mealie.core.root_logger import get_logger 

17from mealie.core.settings.directories import AppDirectories 

18from mealie.core.settings.settings import AppSettings 

19from mealie.db.db_setup import generate_session 

20from mealie.lang import local_provider 

21from mealie.lang.providers import Translator 

22from mealie.repos._utils import NOT_SET, NotSet 

23from mealie.repos.all_repositories import AllRepositories, get_repositories 

24from mealie.routes._base.checks import OperationChecks 

25from mealie.schema.household.household import HouseholdInDB 

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

27from mealie.services.event_bus_service.event_bus_service import EventBusService 

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

29 

30 

31class _BaseController(ABC): # noqa: B024 

32 session: Session = Depends(generate_session) 

33 translator: Translator = Depends(local_provider) 

34 

35 _repos: AllRepositories | None = None 

36 _logger: Logger | None = None 

37 _settings: AppSettings | None = None 

38 _folders: AppDirectories | None = None 

39 

40 @property 

41 def t(self): 

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

43 

44 @property 

45 def repos(self): 

46 if not self._repos: 

47 self._repos = AllRepositories(self.session, group_id=self.group_id, household_id=self.household_id) 

48 return self._repos 

49 

50 @property 

51 def logger(self) -> Logger: 

52 if not self._logger: 

53 self._logger = get_logger() 

54 return self._logger 

55 

56 @property 

57 def settings(self) -> AppSettings: 

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

59 self._settings = get_app_settings() 

60 return self._settings 

61 

62 @property 

63 def folders(self) -> AppDirectories: 

64 if not self._folders: 

65 self._folders = get_app_dirs() 

66 return self._folders 

67 

68 @property 

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

70 return NOT_SET 

71 

72 @property 

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

74 return NOT_SET 

75 

76 model_config = ConfigDict(arbitrary_types_allowed=True) 

77 

78 

79class BasePublicController(_BaseController): 

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 ... 

87 

88 

89class BasePublicGroupExploreController(BasePublicController): 

90 """ 

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

92 """ 

93 

94 group: GroupInDB = Depends(get_public_group) 

95 

96 @property 

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

98 return self.group.id 

99 

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

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: 

107 if endpoint.startswith("/"): 

108 endpoint = endpoint[1:] 

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

110 

111 

112class BasePublicHouseholdExploreController(BasePublicGroupExploreController): 

113 """ 

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

115 """ 

116 

117 @property 

118 def cross_household_repos(self): 

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): 

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) 

138 integration_id: str = Depends(get_integration_id) 

139 translator: Translator = Depends(local_provider) 

140 

141 # Manual Cache 

142 _checks: OperationChecks 

143 

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

145 registered = { 

146 **mealie_registered_exceptions(self.translator), 

147 } 

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

149 

150 @property 

151 def group_id(self) -> UUID4: 

152 return self.user.group_id 

153 

154 @property 

155 def household_id(self) -> UUID4: 

156 return self.user.household_id 

157 

158 @property 

159 def group(self) -> GroupInDB: 

160 return self.repos.groups.get_one(self.group_id) 

161 

162 @property 

163 def household(self) -> HouseholdInDB: 

164 return self.repos.households.get_one(self.household_id) 

165 

166 @property 

167 def checks(self) -> OperationChecks: 

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

169 self._checks = OperationChecks(self.user) 

170 return self._checks 

171 

172 

173class BaseAdminController(BaseUserController): 

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) 

181 

182 @property 

183 def repos(self): 

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): 

191 """ 

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

193 """ 

194 

195 event_bus: EventBusService = Depends(EventBusService.as_dependency) 

196 

197 def publish_event( 

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( 

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 )