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

1from datetime import datetime 

2from typing import TYPE_CHECKING 

3 

4from sqlalchemy import Column, ForeignKey, UniqueConstraint, event 

5from sqlalchemy.engine.base import Connection 

6from sqlalchemy.ext.associationproxy import AssociationProxy, association_proxy 

7from sqlalchemy.orm import Mapped, mapped_column, relationship 

8from sqlalchemy.orm.session import Session 

9 

10from mealie.db.models._model_utils.datetime import NaiveDateTime 

11 

12from .._model_base import BaseMixins, SqlAlchemyBase 

13from .._model_utils.auto_init import auto_init 

14from .._model_utils.guid import GUID 

15 

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

17 from ..recipe import RecipeModel 

18 from .household import Household 

19 

20 

21class HouseholdToRecipe(SqlAlchemyBase, BaseMixins): 

22 __tablename__ = "households_to_recipes" 

23 __table_args__ = (UniqueConstraint("household_id", "recipe_id", name="household_id_recipe_id_key"),) 

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

25 

26 household: Mapped["Household"] = relationship("Household", viewonly=True) 

27 household_id = Column(GUID, ForeignKey("households.id"), index=True, primary_key=True) 

28 recipe: Mapped["RecipeModel"] = relationship("RecipeModel", viewonly=True) 

29 recipe_id = Column(GUID, ForeignKey("recipes.id"), index=True, primary_key=True) 

30 group_id: AssociationProxy[GUID] = association_proxy("household", "group_id") 

31 

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

33 

34 @auto_init() 

35 def __init__(self, **_) -> None: 

36 pass 

37 

38 

39def update_recipe_last_made(session: Session, target: HouseholdToRecipe): 

40 if not target.last_made: 40 ↛ 41line 40 didn't jump to line 41 because the condition on line 40 was never true

41 return 

42 

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

44 

45 recipe = session.query(RecipeModel).filter(RecipeModel.id == target.recipe_id).first() 

46 if not recipe: 46 ↛ 47line 46 didn't jump to line 47 because the condition on line 46 was never true

47 return 

48 

49 recipe.last_made = recipe.last_made or target.last_made 

50 recipe.last_made = max(recipe.last_made, target.last_made) 

51 

52 

53@event.listens_for(HouseholdToRecipe, "after_insert") 

54@event.listens_for(HouseholdToRecipe, "after_update") 

55@event.listens_for(HouseholdToRecipe, "after_delete") 

56def update_recipe_rating_on_insert_or_delete(_, connection: Connection, target: HouseholdToRecipe): 

57 session = Session(bind=connection) 

58 

59 update_recipe_last_made(session, target) 

60 session.commit()