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

1from __future__ import annotations 

2 

3import datetime 

4from numbers import Number 

5from pathlib import Path 

6from typing import Annotated, Any, ClassVar 

7from uuid import uuid4 

8 

9from pydantic import UUID4, BaseModel, ConfigDict, Field, field_validator 

10from pydantic_core.core_schema import ValidationInfo 

11from slugify import slugify 

12from sqlalchemy import Select, desc, func, or_, select, text 

13from sqlalchemy.orm import Session, joinedload, selectinload 

14from sqlalchemy.orm.interfaces import LoaderOption 

15 

16from mealie.core.config import get_app_dirs 

17from mealie.core.exceptions import SlugError 

18from mealie.db.models.users.users import User 

19from mealie.schema._mealie import MealieModel, SearchType 

20from mealie.schema._mealie.mealie_model import UpdatedAtField 

21from mealie.schema.response.pagination import PaginationBase 

22 

23from ...db.models.recipe import ( 

24 IngredientFoodModel, 

25 RecipeComment, 

26 RecipeIngredientModel, 

27 RecipeInstruction, 

28 RecipeModel, 

29) 

30from .recipe_asset import RecipeAsset 

31from .recipe_comments import RecipeCommentOut 

32from .recipe_notes import RecipeNote 

33from .recipe_nutrition import Nutrition 

34from .recipe_settings import RecipeSettings 

35from .recipe_step import RecipeStep 

36 

37app_dirs = get_app_dirs() 

38 

39 

40def create_recipe_slug(name: str, max_length: int = 250) -> str: 

41 """Generate a slug from a recipe name, truncating to a reasonable length. 

42 

43 Args: 

44 name: The recipe name to create a slug from 

45 max_length: Maximum length for the slug (default: 250) 

46 

47 Returns: 

48 A truncated slug string 

49 

50 Raises: 

51 ValueError: If the name cannot be converted to a valid slug 

52 """ 

53 generated_slug = slugify(name) 

54 if not generated_slug: 

55 raise SlugError("Recipe name cannot be empty or contain only special characters") 

56 if len(generated_slug) > max_length: 56 ↛ 57line 56 didn't jump to line 57 because the condition on line 56 was never true

57 generated_slug = generated_slug[:max_length] 

58 return generated_slug 

59 

60 

61class RecipeTag(MealieModel): 

62 id: UUID4 | None = None 

63 group_id: UUID4 | None = None 

64 name: str 

65 slug: str 

66 

67 _searchable_properties: ClassVar[list[str]] = ["name"] 

68 model_config = ConfigDict(from_attributes=True) 

69 

70 

71class RecipeTagPagination(PaginationBase): 

72 items: list[RecipeTag] 

73 

74 

75class RecipeCategory(RecipeTag): 

76 pass 

77 

78 

79class RecipeCategoryPagination(PaginationBase): 

80 items: list[RecipeCategory] 

81 

82 

83class RecipeTool(RecipeTag): 

84 id: UUID4 

85 households_with_tool: list[str] = [] 

86 

87 @field_validator("households_with_tool", mode="before") 

88 def convert_households_to_slugs(cls, v): 

89 if not v: 

90 return [] 

91 

92 try: 

93 return [household.slug for household in v] 

94 except AttributeError: 

95 return v 

96 

97 

98class RecipeToolPagination(PaginationBase): 

99 items: list[RecipeTool] 

100 

101 

102class CreateRecipeBulk(BaseModel): 

103 url: str 

104 categories: list[RecipeCategory] | None = None 

105 tags: list[RecipeTag] | None = None 

106 

107 

108class CreateRecipeByUrlBulk(BaseModel): 

109 imports: list[CreateRecipeBulk] 

110 

111 

112class CreateRecipe(MealieModel): 

113 name: str 

114 

115 

116class RecipeSummary(MealieModel): 

117 id: UUID4 | None = None 

118 _normalize_search: ClassVar[bool] = True 

119 

120 user_id: UUID4 = Field(default_factory=uuid4, validate_default=True) 

121 household_id: UUID4 = Field(default_factory=uuid4, validate_default=True) 

122 group_id: UUID4 = Field(default_factory=uuid4, validate_default=True) 

123 

124 name: str | None = None 

125 slug: Annotated[str, Field(validate_default=True)] = "" 

126 image: Any | None = None 

127 recipe_servings: float = 0 

128 recipe_yield_quantity: float = 0 

129 recipe_yield: str | None = None 

130 

131 total_time: str | None = None 

132 prep_time: str | None = None 

133 cook_time: str | None = None 

134 perform_time: str | None = None 

135 

136 description: str | None = "" 

137 recipe_category: Annotated[list[RecipeCategory] | None, Field(validate_default=True)] | None = [] 

138 tags: Annotated[list[RecipeTag] | None, Field(validate_default=True)] = [] 

139 tools: list[RecipeTool] = [] 

140 rating: float | None = None 

141 org_url: str | None = Field(None, alias="orgURL") 

142 

143 date_added: datetime.date | None = None 

144 date_updated: datetime.datetime | None = None 

145 

146 created_at: datetime.datetime | None = None 

147 updated_at: datetime.datetime | None = UpdatedAtField(None) 

148 last_made: datetime.datetime | None = None 

149 model_config = ConfigDict(from_attributes=True) 

150 

151 @field_validator("recipe_yield", "total_time", "prep_time", "cook_time", "perform_time", mode="before") 

152 def clean_strings(val: Any): 

153 if val is None: 

154 return val 

155 if isinstance(val, Number): 

156 return str(val) 

157 

158 return val 

159 

160 @property 

161 def recipe_yield_display(self) -> str: 

162 return f"{self.recipe_yield_quantity} {self.recipe_yield}".strip() 

163 

164 @classmethod 

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

166 return [ 

167 joinedload(RecipeModel.recipe_category), 

168 joinedload(RecipeModel.tags), 

169 joinedload(RecipeModel.tools), 

170 joinedload(RecipeModel.user).load_only(User.household_id), 

171 ] 

172 

173 

174class RecipePagination(PaginationBase): 

175 items: list[RecipeSummary] 

176 

177 

178class Recipe(RecipeSummary): 

179 recipe_ingredient: Annotated[list[RecipeIngredient], Field(validate_default=True)] = [] 

180 recipe_instructions: list[RecipeStep] | None = [] 

181 nutrition: Nutrition | None = None 

182 

183 # Mealie Specific 

184 settings: RecipeSettings | None = None 

185 assets: list[RecipeAsset] | None = [] 

186 notes: list[RecipeNote] | None = [] 

187 extras: dict | None = {} 

188 

189 comments: list[RecipeCommentOut] | None = [] 

190 

191 @staticmethod 

192 def _get_dir(dir: Path) -> Path: 

193 """Gets a directory and creates it if it doesn't exist""" 

194 

195 dir.mkdir(exist_ok=True, parents=True) 

196 return dir 

197 

198 @classmethod 

199 def directory_from_id(cls, recipe_id: UUID4 | str) -> Path: 

200 return cls._get_dir(app_dirs.RECIPE_DATA_DIR.joinpath(str(recipe_id))) 

201 

202 @classmethod 

203 def asset_dir_from_id(cls, recipe_id: UUID4 | str) -> Path: 

204 return cls._get_dir(cls.directory_from_id(recipe_id).joinpath("assets")) 

205 

206 @classmethod 

207 def image_dir_from_id(cls, recipe_id: UUID4 | str) -> Path: 

208 return cls._get_dir(cls.directory_from_id(recipe_id).joinpath("images")) 

209 

210 @classmethod 

211 def timeline_image_dir_from_id(cls, recipe_id: UUID4 | str, timeline_event_id: UUID4 | str) -> Path: 

212 return cls._get_dir(cls.image_dir_from_id(recipe_id).joinpath("timeline").joinpath(str(timeline_event_id))) 

213 

214 @property 

215 def directory(self) -> Path: 

216 if not self.id: 216 ↛ 217line 216 didn't jump to line 217 because the condition on line 216 was never true

217 raise ValueError("Recipe has no ID") 

218 

219 return self.directory_from_id(self.id) 

220 

221 @property 

222 def asset_dir(self) -> Path: 

223 if not self.id: 223 ↛ 224line 223 didn't jump to line 224 because the condition on line 223 was never true

224 raise ValueError("Recipe has no ID") 

225 

226 return self.asset_dir_from_id(self.id) 

227 

228 @property 

229 def image_dir(self) -> Path: 

230 if not self.id: 

231 raise ValueError("Recipe has no ID") 

232 

233 return self.image_dir_from_id(self.id) 

234 

235 model_config = ConfigDict(from_attributes=True) 

236 

237 @field_validator("slug", mode="before") 

238 def validate_slug(slug: str, info: ValidationInfo): 

239 if not info.data.get("name"): 

240 return slug 

241 

242 return create_recipe_slug(info.data["name"]) 

243 

244 @field_validator("recipe_ingredient", mode="before") 

245 def validate_ingredients(recipe_ingredient): 

246 if not recipe_ingredient or not isinstance(recipe_ingredient, list): 

247 return recipe_ingredient 

248 

249 if all(isinstance(elem, str) for elem in recipe_ingredient): 

250 return [RecipeIngredient(note=x) for x in recipe_ingredient] 

251 

252 return recipe_ingredient 

253 

254 @field_validator("tags", mode="before") 

255 def validate_tags(cats: list[Any]): 

256 if isinstance(cats, list) and cats and isinstance(cats[0], str): 

257 return [RecipeTag(id=uuid4(), name=c, slug=slugify(c)) for c in cats] 

258 return cats 

259 

260 @field_validator("recipe_category", mode="before") 

261 def validate_categories(cats: list[Any]): 

262 if isinstance(cats, list) and cats and isinstance(cats[0], str): 

263 return [RecipeCategory(id=uuid4(), name=c, slug=slugify(c)) for c in cats] 

264 return cats 

265 

266 @field_validator("group_id", mode="before") 

267 def validate_group_id(group_id: Any): 

268 if isinstance(group_id, int): 

269 return uuid4() 

270 return group_id 

271 

272 @field_validator("household_id", mode="before") 

273 def validate_household_id(household_id: Any): 

274 if isinstance(household_id, int): 

275 return uuid4() 

276 return household_id 

277 

278 @field_validator("user_id", mode="before") 

279 def validate_user_id(user_id: Any): 

280 if isinstance(user_id, int): 

281 return uuid4() 

282 return user_id 

283 

284 @field_validator("extras", mode="before") 

285 def convert_extras_to_dict(cls, v): 

286 if isinstance(v, dict): 

287 return v 

288 

289 return {x.key_name: x.value for x in v} if v else {} 

290 

291 @field_validator("nutrition", mode="before") 

292 def validate_nutrition(cls, v): 

293 return v or None 

294 

295 @classmethod 

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

297 return [ 

298 selectinload(RecipeModel.assets), 

299 selectinload(RecipeModel.comments).joinedload(RecipeComment.user), 

300 selectinload(RecipeModel.extras), 

301 joinedload(RecipeModel.recipe_category), 

302 selectinload(RecipeModel.tags), 

303 selectinload(RecipeModel.tools), 

304 selectinload(RecipeModel.recipe_ingredient).joinedload(RecipeIngredientModel.unit), 

305 selectinload(RecipeModel.recipe_ingredient) 

306 .joinedload(RecipeIngredientModel.food) 

307 .joinedload(IngredientFoodModel.extras), 

308 selectinload(RecipeModel.recipe_ingredient) 

309 .joinedload(RecipeIngredientModel.food) 

310 .joinedload(IngredientFoodModel.label), 

311 selectinload(RecipeModel.recipe_instructions).joinedload(RecipeInstruction.ingredient_references), 

312 joinedload(RecipeModel.nutrition), 

313 joinedload(RecipeModel.settings), 

314 # for whatever reason, joinedload can mess up the order here, so use selectinload just this once 

315 selectinload(RecipeModel.notes), 

316 ] 

317 

318 @classmethod 

319 def filter_search_query( 

320 cls, db_model, query: Select, session: Session, search_type: SearchType, search: str, search_list: list[str] 

321 ) -> Select: 

322 """ 

323 1. token search looks for any individual exact hit in name, description, and ingredients 

324 2. fuzzy search looks for trigram hits in name, description, and ingredients 

325 3. Sort order is determined by closeness to the recipe name 

326 Should search also look at tags? 

327 """ 

328 

329 if search_type is SearchType.fuzzy: 329 ↛ 333line 329 didn't jump to line 333 because the condition on line 329 was never true

330 # I would prefer to just do this in the recipe_ingredient.any part of the main query, 

331 # but it turns out that at least sqlite wont use indexes for that correctly anymore and 

332 # takes a big hit, so prefiltering it is 

333 ingredient_ids = ( 

334 session.execute( 

335 select(RecipeIngredientModel.id).filter( 

336 or_( 

337 RecipeIngredientModel.note_normalized.op("%>")(search), 

338 RecipeIngredientModel.original_text_normalized.op("%>")(search), 

339 ) 

340 ) 

341 ) 

342 .scalars() 

343 .all() 

344 ) 

345 

346 session.execute(text(f"set pg_trgm.word_similarity_threshold = {cls._fuzzy_similarity_threshold};")) 

347 return query.filter( 

348 or_( 

349 RecipeModel.name_normalized.op("%>")(search), 

350 RecipeModel.description_normalized.op("%>")(search), 

351 RecipeModel.recipe_ingredient.any(RecipeIngredientModel.id.in_(ingredient_ids)), 

352 ) 

353 ).order_by( # trigram ordering could be too slow on million record db, but is fine with thousands. 

354 func.least( 

355 RecipeModel.name_normalized.op("<->>")(search), 

356 ) 

357 ) 

358 

359 else: 

360 ingredient_ids = ( 

361 session.execute( 

362 select(RecipeIngredientModel.id).filter( 

363 or_( 

364 *[RecipeIngredientModel.note_normalized.like(f"%{ns}%") for ns in search_list], 

365 *[RecipeIngredientModel.original_text_normalized.like(f"%{ns}%") for ns in search_list], 

366 ) 

367 ) 

368 ) 

369 .scalars() 

370 .all() 

371 ) 

372 

373 return query.filter( 

374 or_( 

375 *[RecipeModel.name_normalized.like(f"%{ns}%") for ns in search_list], 

376 *[RecipeModel.description_normalized.like(f"%{ns}%") for ns in search_list], 

377 RecipeModel.recipe_ingredient.any(RecipeIngredientModel.id.in_(ingredient_ids)), 

378 ) 

379 ).order_by(desc(RecipeModel.name_normalized.like(f"%{search}%"))) 

380 

381 

382class RecipeLastMade(BaseModel): 

383 timestamp: datetime.datetime 

384 

385 

386from mealie.schema.recipe.recipe_ingredient import RecipeIngredient # noqa: E402 

387 

388RecipeSummary.model_rebuild() 

389Recipe.model_rebuild()