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

1from datetime import UTC, date, datetime 

2from typing import TYPE_CHECKING 

3 

4import sqlalchemy as sa 

5import sqlalchemy.orm as orm 

6from pydantic import ConfigDict 

7from sqlalchemy import event 

8from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 

9from sqlalchemy.ext.orderinglist import ordering_list 

10from sqlalchemy.orm import Mapped, mapped_column, validates 

11from sqlalchemy.orm.attributes import get_history 

12from sqlalchemy.orm.session import object_session 

13 

14from mealie.db.models._model_utils.auto_init import auto_init 

15from mealie.db.models._model_utils.datetime import NaiveDateTime, get_utc_today 

16from mealie.db.models._model_utils.guid import GUID 

17 

18from .._model_base import BaseMixins, SqlAlchemyBase 

19from ..household.household_to_recipe import HouseholdToRecipe 

20from ..users.user_to_recipe import UserToRecipe 

21from .api_extras import ApiExtras, api_extras 

22from .assets import RecipeAsset 

23from .category import recipes_to_categories 

24from .comment import RecipeComment 

25from .ingredient import RecipeIngredientModel 

26from .instruction import RecipeInstruction 

27from .note import Note 

28from .nutrition import Nutrition 

29from .recipe_timeline import RecipeTimelineEvent 

30from .settings import RecipeSettings 

31from .shared import RecipeShareTokenModel 

32from .tag import recipes_to_tags 

33from .tool import recipes_to_tools 

34 

35if TYPE_CHECKING: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true

36 from ..group import Group, GroupMealPlan 

37 from ..household import Household, ShoppingListItemRecipeReference, ShoppingListRecipeReference 

38 from ..users import User 

39 from . import Category, Tag, Tool 

40 

41 

42class RecipeModel(SqlAlchemyBase, BaseMixins): 

43 __tablename__ = "recipes" 

44 __table_args__: tuple[sa.UniqueConstraint, ...] = ( 

45 sa.UniqueConstraint("slug", "group_id", name="recipe_slug_group_id_key"), 

46 ) 

47 

48 id: Mapped[GUID] = mapped_column(GUID, primary_key=True, default=GUID.generate) 

49 slug: Mapped[str | None] = mapped_column(sa.String, index=True) 

50 

51 # ID Relationships 

52 group_id: Mapped[GUID] = mapped_column(GUID, sa.ForeignKey("groups.id"), nullable=False, index=True) 

53 group: Mapped["Group"] = orm.relationship("Group", back_populates="recipes", foreign_keys=[group_id]) 

54 

55 household_id: AssociationProxy[GUID] = association_proxy("user", "household_id") 

56 household: AssociationProxy["Household"] = association_proxy("user", "household") 

57 

58 user_id: Mapped[GUID | None] = mapped_column(GUID, sa.ForeignKey("users.id", use_alter=True), index=True) 

59 user: Mapped["User"] = orm.relationship("User", uselist=False, foreign_keys=[user_id]) 

60 

61 rating: Mapped[float | None] = mapped_column(sa.Float, index=True, nullable=True) 

62 rated_by: Mapped[list["User"]] = orm.relationship( 

63 "User", 

64 secondary=UserToRecipe.__tablename__, 

65 back_populates="rated_recipes", 

66 overlaps="recipe,favorited_by,favorited_recipes", 

67 ) 

68 favorited_by: Mapped[list["User"]] = orm.relationship( 

69 "User", 

70 secondary=UserToRecipe.__tablename__, 

71 primaryjoin="and_(RecipeModel.id==UserToRecipe.recipe_id, UserToRecipe.is_favorite==True)", 

72 back_populates="favorite_recipes", 

73 overlaps="recipe,rated_by,rated_recipes", 

74 ) 

75 

76 meal_entries: Mapped[list["GroupMealPlan"]] = orm.relationship( 

77 "GroupMealPlan", back_populates="recipe", cascade="all, delete-orphan" 

78 ) 

79 

80 # General Recipe Properties 

81 name: Mapped[str] = mapped_column(sa.String, nullable=False) 

82 description: Mapped[str | None] = mapped_column(sa.String) 

83 

84 image: Mapped[str | None] = mapped_column(sa.String) 

85 

86 # Time Related Properties 

87 total_time: Mapped[str | None] = mapped_column(sa.String) 

88 prep_time: Mapped[str | None] = mapped_column(sa.String) 

89 perform_time: Mapped[str | None] = mapped_column(sa.String) 

90 cook_time: Mapped[str | None] = mapped_column(sa.String) 

91 

92 recipe_yield: Mapped[str | None] = mapped_column(sa.String) 

93 recipe_yield_quantity: Mapped[float] = mapped_column(sa.Float, index=True, default=0) 

94 recipe_servings: Mapped[float] = mapped_column(sa.Float, index=True, default=0) 

95 

96 assets: Mapped[list[RecipeAsset]] = orm.relationship("RecipeAsset", cascade="all, delete-orphan") 

97 nutrition: Mapped[Nutrition] = orm.relationship("Nutrition", uselist=False, cascade="all, delete-orphan") 

98 recipe_category: Mapped[list["Category"]] = orm.relationship( 

99 "Category", secondary=recipes_to_categories, back_populates="recipes" 

100 ) 

101 tools: Mapped[list["Tool"]] = orm.relationship("Tool", secondary=recipes_to_tools, back_populates="recipes") 

102 

103 recipe_ingredient: Mapped[list[RecipeIngredientModel]] = orm.relationship( 

104 "RecipeIngredientModel", 

105 cascade="all, delete-orphan", 

106 order_by="RecipeIngredientModel.position", 

107 collection_class=ordering_list("position"), 

108 ) 

109 recipe_instructions: Mapped[list[RecipeInstruction]] = orm.relationship( 

110 "RecipeInstruction", 

111 cascade="all, delete-orphan", 

112 order_by="RecipeInstruction.position", 

113 collection_class=ordering_list("position"), 

114 ) 

115 

116 share_tokens: Mapped[list[RecipeShareTokenModel]] = orm.relationship( 

117 RecipeShareTokenModel, back_populates="recipe", cascade="all, delete, delete-orphan" 

118 ) 

119 

120 comments: Mapped[list[RecipeComment]] = orm.relationship( 

121 "RecipeComment", back_populates="recipe", cascade="all, delete, delete-orphan" 

122 ) 

123 

124 timeline_events: Mapped[list[RecipeTimelineEvent]] = orm.relationship( 

125 "RecipeTimelineEvent", back_populates="recipe", cascade="all, delete, delete-orphan" 

126 ) 

127 

128 # Mealie Specific 

129 settings: Mapped[list["RecipeSettings"]] = orm.relationship( 

130 "RecipeSettings", uselist=False, cascade="all, delete-orphan" 

131 ) 

132 tags: Mapped[list["Tag"]] = orm.relationship("Tag", secondary=recipes_to_tags, back_populates="recipes") 

133 notes: Mapped[list[Note]] = orm.relationship("Note", cascade="all, delete-orphan") 

134 org_url: Mapped[str | None] = mapped_column(sa.String) 

135 extras: Mapped[list[ApiExtras]] = orm.relationship("ApiExtras", cascade="all, delete-orphan") 

136 

137 # Time Stamp Properties 

138 date_added: Mapped[date | None] = mapped_column(sa.Date, default=get_utc_today) 

139 date_updated: Mapped[datetime | None] = mapped_column(NaiveDateTime) 

140 

141 last_made: Mapped[datetime | None] = mapped_column(NaiveDateTime) 

142 made_by: Mapped[list["Household"]] = orm.relationship( 

143 "Household", secondary=HouseholdToRecipe.__tablename__, back_populates="made_recipes" 

144 ) 

145 

146 # Shopping List Refs 

147 shopping_list_refs: Mapped[list["ShoppingListRecipeReference"]] = orm.relationship( 

148 "ShoppingListRecipeReference", 

149 back_populates="recipe", 

150 cascade="all, delete-orphan", 

151 ) 

152 shopping_list_item_refs: Mapped[list["ShoppingListItemRecipeReference"]] = orm.relationship( 

153 "ShoppingListItemRecipeReference", 

154 back_populates="recipe", 

155 cascade="all, delete-orphan", 

156 ) 

157 

158 # Automatically updated by sqlalchemy event, do not write to this manually 

159 name_normalized: Mapped[str] = mapped_column(sa.String, nullable=False, index=True) 

160 description_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 

161 model_config = ConfigDict( 

162 get_attr="slug", 

163 exclude={ 

164 "assets", 

165 "notes", 

166 "nutrition", 

167 "recipe_ingredient", 

168 "recipe_instructions", 

169 "settings", 

170 "comments", 

171 "timeline_events", 

172 }, 

173 ) 

174 

175 # Deprecated 

176 recipeCuisine: Mapped[str | None] = mapped_column(sa.String) 

177 is_ocr_recipe: Mapped[bool | None] = mapped_column(sa.Boolean, default=False) 

178 

179 @validates("name") 

180 def validate_name(self, _, name): 

181 assert name != "" 

182 return name 

183 

184 @api_extras 

185 @auto_init() 

186 def __init__( 

187 self, 

188 session, 

189 name: str | None = None, 

190 description: str | None = None, 

191 assets: list | None = None, 

192 notes: list[dict] | None = None, 

193 nutrition: dict | None = None, 

194 recipe_ingredient: list[dict] | None = None, 

195 recipe_instructions: list[dict] | None = None, 

196 settings: dict | None = None, 

197 **_, 

198 ) -> None: 

199 self.nutrition = Nutrition(**(nutrition or {})) 

200 

201 if recipe_instructions is not None: 

202 self.recipe_instructions = [RecipeInstruction(**step, session=session) for step in recipe_instructions] 

203 

204 if recipe_ingredient is not None: 204 ↛ 207line 204 didn't jump to line 207 because the condition on line 204 was always true

205 self.recipe_ingredient = [RecipeIngredientModel(**ingr, session=session) for ingr in recipe_ingredient] 

206 

207 if assets: 

208 self.assets = [RecipeAsset(**a) for a in assets] 

209 

210 self.settings = RecipeSettings(**(settings or {})) 

211 

212 if notes: 

213 self.notes = [Note(**n) for n in notes] 

214 

215 self.date_updated = datetime.now(UTC) 

216 

217 # SQLAlchemy events do not seem to register things that are set during auto_init 

218 if name is not None: 218 ↛ 221line 218 didn't jump to line 221 because the condition on line 218 was always true

219 self.name_normalized = self.normalize(name) 

220 

221 if description is not None: 

222 self.description_normalized = self.normalize(description) 

223 

224 tableargs = [ # base set of indices 

225 sa.UniqueConstraint("slug", "group_id", name="recipe_slug_group_id_key"), 

226 sa.Index( 

227 "ix_recipes_name_normalized", 

228 "name_normalized", 

229 unique=False, 

230 ), 

231 sa.Index( 

232 "ix_recipes_description_normalized", 

233 "description_normalized", 

234 unique=False, 

235 ), 

236 ] 

237 

238 if session.get_bind().name == "postgresql": 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true

239 tableargs.extend( 

240 [ 

241 sa.Index( 

242 "ix_recipes_name_normalized_gin", 

243 "name_normalized", 

244 unique=False, 

245 postgresql_using="gin", 

246 postgresql_ops={ 

247 "name_normalized": "gin_trgm_ops", 

248 }, 

249 ), 

250 sa.Index( 

251 "ix_recipes_description_normalized_gin", 

252 "description_normalized", 

253 unique=False, 

254 postgresql_using="gin", 

255 postgresql_ops={ 

256 "description_normalized": "gin_trgm_ops", 

257 }, 

258 ), 

259 ] 

260 ) 

261 # add indices 

262 self.__table_args__ = tuple(tableargs) 

263 

264 

265@event.listens_for(RecipeModel.name, "set") 

266def receive_name(target: RecipeModel, value: str, oldvalue, initiator): 

267 target.name_normalized = RecipeModel.normalize(value) 

268 

269 

270@event.listens_for(RecipeModel.description, "set") 

271def receive_description(target: RecipeModel, value: str, oldvalue, initiator): 

272 if value is not None: 

273 target.description_normalized = RecipeModel.normalize(value) 

274 else: 

275 target.description_normalized = None 

276 

277 

278@event.listens_for(RecipeModel, "before_update") 

279def calculate_rating(mapper, connection, target: RecipeModel): 

280 session = object_session(target) 

281 if not (session and session.is_modified(target, "rating")): 

282 return 

283 

284 history = get_history(target, "rating") 

285 old_value = history.deleted[0] if history.deleted else None 

286 new_value = history.added[0] if history.added else None 

287 if old_value == new_value: 

288 return 

289 

290 target.rating = ( 

291 session.query(sa.func.avg(UserToRecipe.rating)) 

292 .filter(UserToRecipe.recipe_id == target.id, UserToRecipe.rating is not None, UserToRecipe.rating > 0) 

293 .scalar() 

294 )