Coverage for opt/mealie/lib/python3.12/site-packages/mealie/db/models/household/shopping_list.py: 77%

147 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-11-25 15:48 +0000

1from contextvars import ContextVar 1a

2from datetime import UTC, datetime 1a

3from typing import TYPE_CHECKING, Optional 1a

4 

5from pydantic import ConfigDict 1a

6from sqlalchemy import Boolean, Float, ForeignKey, Integer, String, UniqueConstraint, event, orm 1a

7from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 1a

8from sqlalchemy.ext.orderinglist import ordering_list 1a

9from sqlalchemy.orm import Mapped, mapped_column 1a

10 

11from mealie.db.models.labels import MultiPurposeLabel 1a

12from mealie.db.models.recipe.api_extras import ShoppingListExtras, ShoppingListItemExtras, api_extras 1a

13 

14from .._model_base import BaseMixins, SqlAlchemyBase 1a

15from .._model_utils.auto_init import auto_init 1a

16from .._model_utils.guid import GUID 1a

17from ..recipe.ingredient import IngredientFoodModel, IngredientUnitModel 1a

18 

19if TYPE_CHECKING: 19 ↛ 20line 19 didn't jump to line 20 because the condition on line 19 was never true1a

20 from ..group import Group 

21 from ..recipe import RecipeModel 

22 from ..users import User 

23 from .household import Household 

24 

25 

26class ShoppingListItemRecipeReference(BaseMixins, SqlAlchemyBase): 1a

27 __tablename__ = "shopping_list_item_recipe_reference" 1a

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

29 

30 shopping_list_item: Mapped["ShoppingListItem"] = orm.relationship( 1a

31 "ShoppingListItem", back_populates="recipe_references" 

32 ) 

33 shopping_list_item_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("shopping_list_items.id"), primary_key=True) 1a

34 

35 recipe_id: Mapped[GUID | None] = mapped_column(GUID, ForeignKey("recipes.id"), index=True) 1a

36 recipe: Mapped[Optional["RecipeModel"]] = orm.relationship("RecipeModel", back_populates="shopping_list_item_refs") 1a

37 recipe_quantity: Mapped[float] = mapped_column(Float, nullable=False) 1a

38 recipe_scale: Mapped[float] = mapped_column(Float, default=1) 1a

39 recipe_note: Mapped[str | None] = mapped_column(String) 1a

40 

41 group_id: AssociationProxy[GUID] = association_proxy("shopping_list_item", "group_id") 1a

42 household_id: AssociationProxy[GUID] = association_proxy("shopping_list_item", "household_id") 1a

43 

44 @auto_init() 1a

45 def __init__(self, **_) -> None: 1a

46 pass 

47 

48 

49class ShoppingListItem(SqlAlchemyBase, BaseMixins): 1a

50 __tablename__ = "shopping_list_items" 1a

51 

52 # Id's 

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

54 shopping_list: Mapped["ShoppingList"] = orm.relationship("ShoppingList", back_populates="list_items") 1a

55 shopping_list_id: Mapped[GUID | None] = mapped_column(GUID, ForeignKey("shopping_lists.id"), index=True) 1a

56 

57 group_id: AssociationProxy[GUID] = association_proxy("shopping_list", "group_id") 1a

58 household_id: AssociationProxy[GUID] = association_proxy("shopping_list", "household_id") 1a

59 

60 # Meta 

61 is_ingredient: Mapped[bool | None] = mapped_column(Boolean, default=True) 1a

62 position: Mapped[int] = mapped_column(Integer, nullable=False, default=0, index=True) 1a

63 checked: Mapped[bool | None] = mapped_column(Boolean, default=False) 1a

64 

65 quantity: Mapped[float | None] = mapped_column(Float, default=1) 1a

66 note: Mapped[str | None] = mapped_column(String) 1a

67 

68 extras: Mapped[list[ShoppingListItemExtras]] = orm.relationship( 1a

69 "ShoppingListItemExtras", cascade="all, delete-orphan" 

70 ) 

71 

72 # Scaling Items 

73 unit_id: Mapped[GUID | None] = mapped_column(GUID, ForeignKey("ingredient_units.id")) 1a

74 unit: Mapped[IngredientUnitModel | None] = orm.relationship(IngredientUnitModel, uselist=False) 1a

75 

76 food_id: Mapped[GUID | None] = mapped_column(GUID, ForeignKey("ingredient_foods.id")) 1a

77 food: Mapped[IngredientFoodModel | None] = orm.relationship(IngredientFoodModel, uselist=False) 1a

78 

79 label_id: Mapped[GUID | None] = mapped_column(GUID, ForeignKey("multi_purpose_labels.id")) 1a

80 label: Mapped[MultiPurposeLabel | None] = orm.relationship( 1a

81 MultiPurposeLabel, uselist=False, back_populates="shopping_list_items" 

82 ) 

83 

84 # Recipe Reference 

85 recipe_references: Mapped[list[ShoppingListItemRecipeReference]] = orm.relationship( 1a

86 ShoppingListItemRecipeReference, cascade="all, delete, delete-orphan" 

87 ) 

88 model_config = ConfigDict(exclude={"label", "food", "unit"}) 1a

89 

90 # Deprecated 

91 is_food: Mapped[bool | None] = mapped_column(Boolean, default=False) 1a

92 

93 @api_extras 1a

94 @auto_init() 1a

95 def __init__(self, **_) -> None: 1a

96 pass 

97 

98 

99class ShoppingListRecipeReference(BaseMixins, SqlAlchemyBase): 1a

100 __tablename__ = "shopping_list_recipe_reference" 1a

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

102 

103 shopping_list: Mapped["ShoppingList"] = orm.relationship("ShoppingList", back_populates="recipe_references") 1a

104 shopping_list_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("shopping_lists.id"), primary_key=True) 1a

105 group_id: AssociationProxy[GUID] = association_proxy("shopping_list", "group_id") 1a

106 household_id: AssociationProxy[GUID] = association_proxy("shopping_list", "household_id") 1a

107 

108 recipe_id: Mapped[GUID | None] = mapped_column(GUID, ForeignKey("recipes.id"), index=True) 1a

109 recipe: Mapped[Optional["RecipeModel"]] = orm.relationship( 1a

110 "RecipeModel", uselist=False, back_populates="shopping_list_refs" 

111 ) 

112 

113 recipe_quantity: Mapped[float] = mapped_column(Float, nullable=False) 1a

114 model_config = ConfigDict(exclude={"id", "recipe"}) 1a

115 

116 @auto_init() 1a

117 def __init__(self, **_) -> None: 1a

118 pass 

119 

120 

121class ShoppingListMultiPurposeLabel(SqlAlchemyBase, BaseMixins): 1a

122 __tablename__ = "shopping_lists_multi_purpose_labels" 1a

123 __table_args__ = (UniqueConstraint("shopping_list_id", "label_id", name="shopping_list_id_label_id_key"),) 1a

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

125 

126 shopping_list_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("shopping_lists.id"), primary_key=True) 1a

127 shopping_list: Mapped["ShoppingList"] = orm.relationship("ShoppingList", back_populates="label_settings") 1a

128 

129 label_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("multi_purpose_labels.id"), primary_key=True) 1a

130 label: Mapped["MultiPurposeLabel"] = orm.relationship( 1a

131 "MultiPurposeLabel", back_populates="shopping_lists_label_settings" 

132 ) 

133 

134 group_id: AssociationProxy[GUID] = association_proxy("shopping_list", "group_id") 1a

135 household_id: AssociationProxy[GUID] = association_proxy("shopping_list", "household_id") 1a

136 

137 position: Mapped[int] = mapped_column(Integer, nullable=False, default=0) 1a

138 model_config = ConfigDict(exclude={"label"}) 1a

139 

140 @auto_init() 1a

141 def __init__(self, **_) -> None: 1a

142 pass 1bcdefghpijklmno

143 

144 

145class ShoppingList(SqlAlchemyBase, BaseMixins): 1a

146 __tablename__ = "shopping_lists" 1a

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

148 

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

150 group: Mapped["Group"] = orm.relationship("Group", back_populates="shopping_lists") 1a

151 household_id: AssociationProxy[GUID] = association_proxy("user", "household_id") 1a

152 household: AssociationProxy["Household"] = association_proxy("user", "household") 1a

153 user_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("users.id"), nullable=False, index=True) 1a

154 user: Mapped["User"] = orm.relationship("User", back_populates="shopping_lists") 1a

155 

156 name: Mapped[str | None] = mapped_column(String) 1a

157 list_items: Mapped[list[ShoppingListItem]] = orm.relationship( 1a

158 ShoppingListItem, 

159 cascade="all, delete, delete-orphan", 

160 order_by="ShoppingListItem.position", 

161 collection_class=ordering_list("position"), 

162 ) 

163 

164 recipe_references: Mapped[list[ShoppingListRecipeReference]] = orm.relationship( 1a

165 ShoppingListRecipeReference, cascade="all, delete, delete-orphan" 

166 ) 

167 label_settings: Mapped[list["ShoppingListMultiPurposeLabel"]] = orm.relationship( 1a

168 ShoppingListMultiPurposeLabel, 

169 cascade="all, delete, delete-orphan", 

170 order_by="ShoppingListMultiPurposeLabel.position", 

171 collection_class=ordering_list("position"), 

172 ) 

173 extras: Mapped[list[ShoppingListExtras]] = orm.relationship("ShoppingListExtras", cascade="all, delete-orphan") 1a

174 model_config = ConfigDict(exclude={"id", "list_items"}) 1a

175 

176 @api_extras 1a

177 @auto_init() 1a

178 def __init__(self, **_) -> None: 1a

179 pass 1bcdefghijklmno

180 

181 

182class SessionBuffer: 1a

183 def __init__(self) -> None: 1a

184 self.shopping_list_ids: set[GUID] = set() 1a

185 

186 def add(self, shopping_list_id: GUID) -> None: 1a

187 self.shopping_list_ids.add(shopping_list_id) 

188 

189 def pop(self) -> GUID | None: 1a

190 try: 

191 return self.shopping_list_ids.pop() 

192 except KeyError: 

193 return None 

194 

195 def clear(self) -> None: 1a

196 self.shopping_list_ids.clear() 

197 

198 

199session_buffer_context = ContextVar("session_buffer", default=SessionBuffer()) # noqa: B039 1a

200 

201 

202@event.listens_for(ShoppingListItem, "after_insert") 1a

203@event.listens_for(ShoppingListItem, "after_update") 1a

204@event.listens_for(ShoppingListItem, "after_delete") 1a

205def buffer_shopping_list_updates(_, connection, target: ShoppingListItem): 1a

206 """Adds the shopping list id to the session buffer so its `updated_at` property can be updated later""" 

207 

208 session_buffer = session_buffer_context.get() 

209 session_buffer.add(target.shopping_list_id) 

210 

211 

212@event.listens_for(orm.Session, "after_flush") 1a

213def update_shopping_lists(session: orm.Session, _): 1a

214 """Pulls all pending shopping list updates from the buffer and updates their `updated_at` property""" 

215 

216 session_buffer = session_buffer_context.get() 1aqrstuvwbcdefghpijklmno

217 if not session_buffer.shopping_list_ids: 217 ↛ 220line 217 didn't jump to line 220 because the condition on line 217 was always true1aqrstuvwbcdefghpijklmno

218 return 1aqrstuvwbcdefghpijklmno

219 

220 local_session = orm.Session(bind=session.connection()) 

221 try: 

222 local_session.begin() 

223 while True: 

224 shopping_list_id = session_buffer.pop() 

225 if not shopping_list_id: 

226 break 

227 

228 shopping_list = local_session.query(ShoppingList).filter(ShoppingList.id == shopping_list_id).first() 

229 if not shopping_list: 

230 continue 

231 

232 shopping_list.updated_at = datetime.now(UTC) 

233 local_session.commit() 

234 except Exception: 

235 local_session.rollback() 

236 raise