Coverage for opt/mealie/lib/python3.12/site-packages/mealie/db/models/recipe/recipe.py: 88%
130 statements
« prev ^ index » next coverage.py v7.10.6, created at 2025-11-25 15:48 +0000
1from datetime import UTC, date, datetime 1a
2from typing import TYPE_CHECKING 1a
4import sqlalchemy as sa 1a
5import sqlalchemy.orm as orm 1a
6from pydantic import ConfigDict 1a
7from sqlalchemy import event 1a
8from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 1a
9from sqlalchemy.ext.orderinglist import ordering_list 1a
10from sqlalchemy.orm import Mapped, mapped_column, validates 1a
11from sqlalchemy.orm.attributes import get_history 1a
12from sqlalchemy.orm.session import object_session 1a
14from mealie.db.models._model_utils.auto_init import auto_init 1a
15from mealie.db.models._model_utils.datetime import NaiveDateTime, get_utc_today 1a
16from mealie.db.models._model_utils.guid import GUID 1a
18from .._model_base import BaseMixins, SqlAlchemyBase 1a
19from ..household.household_to_recipe import HouseholdToRecipe 1a
20from ..users.user_to_recipe import UserToRecipe 1a
21from .api_extras import ApiExtras, api_extras 1a
22from .assets import RecipeAsset 1a
23from .category import recipes_to_categories 1a
24from .comment import RecipeComment 1a
25from .ingredient import RecipeIngredientModel 1a
26from .instruction import RecipeInstruction 1a
27from .note import Note 1a
28from .nutrition import Nutrition 1a
29from .recipe_timeline import RecipeTimelineEvent 1a
30from .settings import RecipeSettings 1a
31from .shared import RecipeShareTokenModel 1a
32from .tag import recipes_to_tags 1a
33from .tool import recipes_to_tools 1a
35if TYPE_CHECKING: 35 ↛ 36line 35 didn't jump to line 36 because the condition on line 35 was never true1a
36 from ..group import Group, GroupMealPlan
37 from ..household import Household, ShoppingListItemRecipeReference, ShoppingListRecipeReference
38 from ..users import User
39 from . import Category, Tag, Tool
42class RecipeModel(SqlAlchemyBase, BaseMixins): 1a
43 __tablename__ = "recipes" 1a
44 __table_args__: tuple[sa.UniqueConstraint, ...] = ( 1a
45 sa.UniqueConstraint("slug", "group_id", name="recipe_slug_group_id_key"),
46 )
48 id: Mapped[GUID] = mapped_column(GUID, primary_key=True, default=GUID.generate) 1a
49 slug: Mapped[str | None] = mapped_column(sa.String, index=True) 1a
51 # ID Relationships
52 group_id: Mapped[GUID] = mapped_column(GUID, sa.ForeignKey("groups.id"), nullable=False, index=True) 1a
53 group: Mapped["Group"] = orm.relationship("Group", back_populates="recipes", foreign_keys=[group_id]) 1a
55 household_id: AssociationProxy[GUID] = association_proxy("user", "household_id") 1a
56 household: AssociationProxy["Household"] = association_proxy("user", "household") 1a
58 user_id: Mapped[GUID | None] = mapped_column(GUID, sa.ForeignKey("users.id", use_alter=True), index=True) 1a
59 user: Mapped["User"] = orm.relationship("User", uselist=False, foreign_keys=[user_id]) 1a
61 rating: Mapped[float | None] = mapped_column(sa.Float, index=True, nullable=True) 1a
62 rated_by: Mapped[list["User"]] = orm.relationship( 1a
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( 1a
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 )
76 meal_entries: Mapped[list["GroupMealPlan"]] = orm.relationship( 1a
77 "GroupMealPlan", back_populates="recipe", cascade="all, delete-orphan"
78 )
80 # General Recipe Properties
81 name: Mapped[str] = mapped_column(sa.String, nullable=False) 1a
82 description: Mapped[str | None] = mapped_column(sa.String) 1a
84 image: Mapped[str | None] = mapped_column(sa.String) 1a
86 # Time Related Properties
87 total_time: Mapped[str | None] = mapped_column(sa.String) 1a
88 prep_time: Mapped[str | None] = mapped_column(sa.String) 1a
89 perform_time: Mapped[str | None] = mapped_column(sa.String) 1a
90 cook_time: Mapped[str | None] = mapped_column(sa.String) 1a
92 recipe_yield: Mapped[str | None] = mapped_column(sa.String) 1a
93 recipe_yield_quantity: Mapped[float] = mapped_column(sa.Float, index=True, default=0) 1a
94 recipe_servings: Mapped[float] = mapped_column(sa.Float, index=True, default=0) 1a
96 assets: Mapped[list[RecipeAsset]] = orm.relationship("RecipeAsset", cascade="all, delete-orphan") 1a
97 nutrition: Mapped[Nutrition] = orm.relationship("Nutrition", uselist=False, cascade="all, delete-orphan") 1a
98 recipe_category: Mapped[list["Category"]] = orm.relationship( 1a
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") 1a
103 recipe_ingredient: Mapped[list[RecipeIngredientModel]] = orm.relationship( 1a
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( 1a
110 "RecipeInstruction",
111 cascade="all, delete-orphan",
112 order_by="RecipeInstruction.position",
113 collection_class=ordering_list("position"),
114 )
116 share_tokens: Mapped[list[RecipeShareTokenModel]] = orm.relationship( 1a
117 RecipeShareTokenModel, back_populates="recipe", cascade="all, delete, delete-orphan"
118 )
120 comments: Mapped[list[RecipeComment]] = orm.relationship( 1a
121 "RecipeComment", back_populates="recipe", cascade="all, delete, delete-orphan"
122 )
124 timeline_events: Mapped[list[RecipeTimelineEvent]] = orm.relationship( 1a
125 "RecipeTimelineEvent", back_populates="recipe", cascade="all, delete, delete-orphan"
126 )
128 # Mealie Specific
129 settings: Mapped[list["RecipeSettings"]] = orm.relationship( 1a
130 "RecipeSettings", uselist=False, cascade="all, delete-orphan"
131 )
132 tags: Mapped[list["Tag"]] = orm.relationship("Tag", secondary=recipes_to_tags, back_populates="recipes") 1a
133 notes: Mapped[list[Note]] = orm.relationship("Note", cascade="all, delete-orphan") 1a
134 org_url: Mapped[str | None] = mapped_column(sa.String) 1a
135 extras: Mapped[list[ApiExtras]] = orm.relationship("ApiExtras", cascade="all, delete-orphan") 1a
137 # Time Stamp Properties
138 date_added: Mapped[date | None] = mapped_column(sa.Date, default=get_utc_today) 1a
139 date_updated: Mapped[datetime | None] = mapped_column(NaiveDateTime) 1a
141 last_made: Mapped[datetime | None] = mapped_column(NaiveDateTime) 1a
142 made_by: Mapped[list["Household"]] = orm.relationship( 1a
143 "Household", secondary=HouseholdToRecipe.__tablename__, back_populates="made_recipes"
144 )
146 # Shopping List Refs
147 shopping_list_refs: Mapped[list["ShoppingListRecipeReference"]] = orm.relationship( 1a
148 "ShoppingListRecipeReference",
149 back_populates="recipe",
150 cascade="all, delete-orphan",
151 )
152 shopping_list_item_refs: Mapped[list["ShoppingListItemRecipeReference"]] = orm.relationship( 1a
153 "ShoppingListItemRecipeReference",
154 back_populates="recipe",
155 cascade="all, delete-orphan",
156 )
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) 1a
160 description_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a
161 model_config = ConfigDict( 1a
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 )
175 # Deprecated
176 recipeCuisine: Mapped[str | None] = mapped_column(sa.String) 1a
177 is_ocr_recipe: Mapped[bool | None] = mapped_column(sa.Boolean, default=False) 1a
179 @validates("name") 1a
180 def validate_name(self, _, name): 1a
181 assert name != "" 1bc
182 return name 1bc
184 @api_extras 1a
185 @auto_init() 1a
186 def __init__( 1a
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 {})) 1bc
201 if recipe_instructions is not None: 201 ↛ 204line 201 didn't jump to line 204 because the condition on line 201 was always true1bc
202 self.recipe_instructions = [RecipeInstruction(**step, session=session) for step in recipe_instructions] 1bc
204 if recipe_ingredient is not None: 204 ↛ 207line 204 didn't jump to line 207 because the condition on line 204 was always true1bc
205 self.recipe_ingredient = [RecipeIngredientModel(**ingr, session=session) for ingr in recipe_ingredient] 1bc
207 if assets: 1bc
208 self.assets = [RecipeAsset(**a) for a in assets] 1bc
210 self.settings = RecipeSettings(**(settings or {})) 1bc
212 if notes: 212 ↛ 213line 212 didn't jump to line 213 because the condition on line 212 was never true1bc
213 self.notes = [Note(**n) for n in notes]
215 self.date_updated = datetime.now(UTC) 1bc
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 true1bc
219 self.name_normalized = self.normalize(name) 1bc
221 if description is not None: 221 ↛ 224line 221 didn't jump to line 224 because the condition on line 221 was always true1bc
222 self.description_normalized = self.normalize(description) 1bc
224 tableargs = [ # base set of indices 1bc
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 ]
238 if session.get_bind().name == "postgresql": 238 ↛ 239line 238 didn't jump to line 239 because the condition on line 238 was never true1bc
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) 1bc
265@event.listens_for(RecipeModel.name, "set") 1a
266def receive_name(target: RecipeModel, value: str, oldvalue, initiator): 1a
267 target.name_normalized = RecipeModel.normalize(value) 1bc
270@event.listens_for(RecipeModel.description, "set") 1a
271def receive_description(target: RecipeModel, value: str, oldvalue, initiator): 1a
272 if value is not None: 272 ↛ 275line 272 didn't jump to line 275 because the condition on line 272 was always true1bc
273 target.description_normalized = RecipeModel.normalize(value) 1bc
274 else:
275 target.description_normalized = None
278@event.listens_for(RecipeModel, "before_update") 1a
279def calculate_rating(mapper, connection, target: RecipeModel): 1a
280 session = object_session(target) 1bc
281 if not (session and session.is_modified(target, "rating")): 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true1bc
282 return
284 history = get_history(target, "rating") 1bc
285 old_value = history.deleted[0] if history.deleted else None 1bc
286 new_value = history.added[0] if history.added else None 1bc
287 if old_value == new_value: 287 ↛ 290line 287 didn't jump to line 290 because the condition on line 287 was always true1bc
288 return 1bc
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 )