Coverage for opt/mealie/lib/python3.12/site-packages/mealie/schema/user/user.py: 96%

1from datetime import UTC, datetime, timedelta 

2from pathlib import Path 

3from typing import Annotated, Any 

4from uuid import UUID 

5 

6from pydantic import UUID4, BaseModel, ConfigDict, Field, StringConstraints, field_validator 

7from sqlalchemy.orm import joinedload, selectinload 

8from sqlalchemy.orm.interfaces import LoaderOption 

9 

10from mealie.core.config import get_app_dirs, get_app_settings 

11from mealie.db.models.recipe.recipe import RecipeModel 

12from mealie.db.models.users import User 

13from mealie.db.models.users.user_to_recipe import UserToRecipe 

14from mealie.db.models.users.users import AuthMethod, LongLiveToken 

15from mealie.schema._mealie import MealieModel 

16from mealie.schema.group.group_preferences import ReadGroupPreferences 

17from mealie.schema.household.webhook import CreateWebhook, ReadWebhook 

18from mealie.schema.response.pagination import PaginationBase 

19 

20from ...db.models.group import Group 

21from ..recipe import CategoryBase 

22 

23DEFAULT_INTEGRATION_ID = "generic" 

24settings = get_app_settings() 

25 

26 

27class LongLiveTokenIn(MealieModel): 

28 name: str 

29 integration_id: str = DEFAULT_INTEGRATION_ID 

30 

31 

32class LongLiveTokenOut(MealieModel): 

33 name: str 

34 id: int 

35 created_at: datetime | None = None 

36 model_config = ConfigDict(from_attributes=True) 

37 

38 @classmethod 

39 def loader_options(cls) -> list[LoaderOption]: 

40 return [joinedload(LongLiveToken.user)] 

41 

42 

43class LongLiveTokenCreateResponse(LongLiveTokenOut): 

44 """Should ONLY be used when creating a new token, as the token field is sensitive""" 

45 

46 token: str 

47 

48 

49class CreateToken(LongLiveTokenIn): 

50 user_id: UUID4 

51 token: str 

52 model_config = ConfigDict(from_attributes=True) 

53 

54 

55class DeleteTokenResponse(MealieModel): 

56 token_delete: str 

57 model_config = ConfigDict(from_attributes=True) 

58 

59 

60class ChangePassword(MealieModel): 

61 current_password: str = "" 

62 new_password: str = Field(..., min_length=8) 

63 

64 

65class GroupBase(MealieModel): 

66 name: Annotated[str, StringConstraints(strip_whitespace=True, min_length=1)] 

67 model_config = ConfigDict(from_attributes=True) 

68 

69 

70class UserRatingSummary(MealieModel): 

71 recipe_id: UUID4 

72 rating: float | None = None 

73 is_favorite: Annotated[bool, Field(validate_default=True)] = False 

74 

75 model_config = ConfigDict(from_attributes=True) 

76 

77 @field_validator("is_favorite", mode="before") 

78 def convert_is_favorite(cls, v: Any) -> bool: 

79 if v is None: 79 ↛ 80line 79 didn't jump to line 80 because the condition on line 79 was never true

80 return False 

81 else: 

82 return v 

83 

84 

85class UserRatingCreate(UserRatingSummary): 

86 user_id: UUID4 

87 

88 

89class UserRatingUpdate(MealieModel): 

90 rating: float | None = None 

91 is_favorite: bool | None = None 

92 

93 

94class UserRatingOut(UserRatingCreate): 

95 id: UUID4 

96 

97 @classmethod 

98 def loader_options(cls) -> list[LoaderOption]: 

99 return [ 

100 joinedload(UserToRecipe.recipe).joinedload(RecipeModel.user).load_only(User.household_id, User.group_id) 

101 ] 

102 

103 

104class UserRatings[DataT: BaseModel](BaseModel): 

105 ratings: list[DataT] 

106 

107 

108class UserBase(MealieModel): 

109 id: UUID4 | None = None 

110 username: str | None = None 

111 full_name: str | None = None 

112 email: Annotated[str, StringConstraints(to_lower=True, strip_whitespace=True)] 

113 auth_method: AuthMethod = AuthMethod.MEALIE 

114 admin: bool = False 

115 group: str | None = None 

116 household: str | None = None 

117 advanced: bool = False 

118 

119 can_invite: bool = False 

120 can_manage: bool = False 

121 can_manage_household: bool = False 

122 can_organize: bool = False 

123 model_config = ConfigDict( 

124 from_attributes=True, 

125 json_schema_extra={ 

126 "example": { 

127 "username": "ChangeMe", 

128 "fullName": "Change Me", 

129 "email": "changeme@example.com", 

130 "group": settings.DEFAULT_GROUP, 

131 "household": settings.DEFAULT_HOUSEHOLD, 

132 "admin": "false", 

133 } 

134 }, 

135 ) 

136 

137 @field_validator("group", mode="before") 

138 def convert_group_to_name(cls, v): 

139 if not v or isinstance(v, str): 

140 return v 

141 

142 try: 

143 return v.name 

144 except AttributeError: 

145 return v 

146 

147 @field_validator("household", mode="before") 

148 def convert_household_to_name(cls, v): 

149 if not v or isinstance(v, str): 

150 return v 

151 

152 try: 

153 return v.name 

154 except AttributeError: 

155 return v 

156 

157 

158class UserIn(UserBase): 

159 username: str 

160 full_name: str 

161 password: str 

162 

163 

164class UserOut(UserBase): 

165 id: UUID4 

166 group: str 

167 group_id: UUID4 

168 group_slug: str 

169 household: str 

170 household_id: UUID4 

171 household_slug: str 

172 tokens: list[LongLiveTokenOut] | None = None 

173 cache_key: str 

174 model_config = ConfigDict(from_attributes=True) 

175 

176 @property 

177 def is_default_user(self) -> bool: 

178 return self.email == settings._DEFAULT_EMAIL.strip().lower() 

179 

180 @classmethod 

181 def loader_options(cls) -> list[LoaderOption]: 

182 return [joinedload(User.group), joinedload(User.household), joinedload(User.tokens)] 

183 

184 

185class UserSummary(MealieModel): 

186 id: UUID4 

187 group_id: UUID4 

188 household_id: UUID4 

189 username: str 

190 full_name: str 

191 model_config = ConfigDict(from_attributes=True) 

192 

193 

194class UserPagination(PaginationBase): 

195 items: list[UserOut] 

196 

197 

198class UserSummaryPagination(PaginationBase): 

199 items: list[UserSummary] 

200 

201 

202class PrivateUser(UserOut): 

203 password: str 

204 login_attemps: int = 0 

205 locked_at: datetime | None = None 

206 model_config = ConfigDict(from_attributes=True) 

207 

208 @field_validator("login_attemps", mode="before") 

209 @classmethod 

210 def none_to_zero(cls, v): 

211 return 0 if v is None else v 

212 

213 @staticmethod 

214 def get_directory(user_id: UUID4 | str) -> Path: 

215 user_dir = get_app_dirs().USER_DIR / str(user_id) 

216 user_dir.mkdir(parents=True, exist_ok=True) 

217 return user_dir 

218 

219 @property 

220 def is_locked(self) -> bool: 

221 if self.locked_at is None: 221 ↛ 224line 221 didn't jump to line 224 because the condition on line 221 was always true

222 return False 

223 

224 lockout_expires_at = self.locked_at + timedelta(hours=get_app_settings().SECURITY_USER_LOCKOUT_TIME) 

225 return lockout_expires_at > datetime.now(UTC) 

226 

227 def directory(self) -> Path: 

228 return PrivateUser.get_directory(self.id) 

229 

230 @classmethod 

231 def loader_options(cls) -> list[LoaderOption]: 

232 return [joinedload(User.group), joinedload(User.household), joinedload(User.tokens)] 

233 

234 

235class UpdateGroup(GroupBase): 

236 id: UUID4 

237 name: str 

238 slug: str 

239 categories: list[CategoryBase] | None = [] 

240 

241 webhooks: list[CreateWebhook] = [] 

242 

243 

244class GroupHouseholdSummary(MealieModel): 

245 id: UUID4 

246 name: str 

247 model_config = ConfigDict(from_attributes=True) 

248 

249 

250class GroupInDB(UpdateGroup): 

251 households: list[GroupHouseholdSummary] | None = None 

252 users: list[UserSummary] | None = None 

253 preferences: ReadGroupPreferences | None = None 

254 webhooks: list[ReadWebhook] = [] 

255 

256 model_config = ConfigDict(from_attributes=True) 

257 

258 @staticmethod 

259 def get_directory(id: UUID4) -> Path: 

260 group_dir = get_app_dirs().GROUPS_DIR / str(id) 

261 group_dir.mkdir(parents=True, exist_ok=True) 

262 return group_dir 

263 

264 @staticmethod 

265 def get_export_directory(id: UUID) -> Path: 

266 export_dir = GroupInDB.get_directory(id) / "export" 

267 export_dir.mkdir(parents=True, exist_ok=True) 

268 return export_dir 

269 

270 @property 

271 def directory(self) -> Path: 

272 return GroupInDB.get_directory(self.id) 

273 

274 @property 

275 def exports(self) -> Path: 

276 return GroupInDB.get_export_directory(self.id) 

277 

278 @classmethod 

279 def loader_options(cls) -> list[LoaderOption]: 

280 return [ 

281 joinedload(Group.categories), 

282 joinedload(Group.webhooks), 

283 joinedload(Group.preferences), 

284 joinedload(Group.households), 

285 selectinload(Group.users).joinedload(User.group), 

286 selectinload(Group.users).joinedload(User.tokens), 

287 ] 

288 

289 

290class GroupSummary(GroupBase): 

291 id: UUID4 

292 name: str 

293 slug: str 

294 preferences: ReadGroupPreferences | None = None 

295 

296 @classmethod 

297 def loader_options(cls) -> list[LoaderOption]: 

298 return [ 

299 joinedload(Group.preferences), 

300 ] 

301 

302 

303class GroupPagination(PaginationBase): 

304 items: list[GroupInDB] 

305 

306 

307class LongLiveTokenInDB(CreateToken): 

308 id: int 

309 user: PrivateUser 

310 model_config = ConfigDict(from_attributes=True)