Coverage for opt/mealie/lib/python3.12/site-packages/mealie/alembic/versions/2024-11-20-17.30.41_b9e516e2d3b3_add_household_to_recipe_last_made_.py: 63%

108 statements  

« prev     ^ index     » next       coverage.py v7.10.6, created at 2025-12-05 13:45 +0000

1"""add household to recipe last made, household to foods, and household to tools 

2 

3Revision ID: b9e516e2d3b3 

4Revises: b1020f328e98 

5Create Date: 2024-11-20 17:30:41.152332 

6 

7""" 

8 

9from datetime import datetime 1a

10 

11import sqlalchemy as sa 1a

12from sqlalchemy import orm 1a

13from sqlalchemy.orm import DeclarativeBase 1a

14 

15import mealie.db.migration_types 1a

16from alembic import op 1a

17from mealie.core.root_logger import get_logger 1a

18from mealie.db.models._model_utils.datetime import NaiveDateTime 1a

19from mealie.db.models._model_utils.guid import GUID 1a

20 

21# revision identifiers, used by Alembic. 

22revision = "b9e516e2d3b3" 1a

23down_revision: str | None = "b1020f328e98" 1a

24branch_labels: str | tuple[str, ...] | None = None 1a

25depends_on: str | tuple[str, ...] | None = None 1a

26 

27logger = get_logger() 1a

28 

29 

30class SqlAlchemyBase(DeclarativeBase): 1a

31 pass 1a

32 

33 

34# Intermediate table definitions 

35class Group(SqlAlchemyBase): 1a

36 __tablename__ = "groups" 1a

37 

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

39 

40 

41class Household(SqlAlchemyBase): 1a

42 __tablename__ = "households" 1a

43 

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

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

46 

47 

48class RecipeModel(SqlAlchemyBase): 1a

49 __tablename__ = "recipes" 1a

50 

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

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

53 last_made: orm.Mapped[datetime | None] = orm.mapped_column(NaiveDateTime) 1a

54 

55 

56class HouseholdToRecipe(SqlAlchemyBase): 1a

57 __tablename__ = "households_to_recipes" 1a

58 

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

60 household_id = sa.Column(GUID, sa.ForeignKey("households.id"), index=True, primary_key=True) 1a

61 recipe_id = sa.Column(GUID, sa.ForeignKey("recipes.id"), index=True, primary_key=True) 1a

62 last_made: orm.Mapped[datetime | None] = orm.mapped_column(NaiveDateTime) 1a

63 

64 

65class IngredientFoodModel(SqlAlchemyBase): 1a

66 __tablename__ = "ingredient_foods" 1a

67 

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

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

70 on_hand: orm.Mapped[bool] = orm.mapped_column(sa.Boolean, default=False) 1a

71 

72 

73class Tool(SqlAlchemyBase): 1a

74 __tablename__ = "tools" 1a

75 

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

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

78 on_hand: orm.Mapped[bool | None] = orm.mapped_column(sa.Boolean, default=False) 1a

79 

80 

81def migrate_recipe_last_made_to_household(session: orm.Session): 1a

82 for group in session.query(Group).all(): 82 ↛ 83line 82 didn't jump to line 83 because the loop on line 82 never started1a

83 households = session.query(Household).filter(Household.group_id == group.id).all() 

84 recipes = ( 

85 session.query(RecipeModel) 

86 .filter( 

87 RecipeModel.group_id == group.id, 

88 RecipeModel.last_made != None, # noqa E711 

89 ) 

90 .all() 

91 ) 

92 

93 for recipe in recipes: 

94 for household in households: 

95 session.add( 

96 HouseholdToRecipe( 

97 household_id=household.id, 

98 recipe_id=recipe.id, 

99 last_made=recipe.last_made, 

100 ) 

101 ) 

102 

103 

104def migrate_foods_on_hand_to_household(session: orm.Session): 1a

105 dialect = op.get_bind().dialect 1a

106 

107 for group in session.query(Group).all(): 107 ↛ 108line 107 didn't jump to line 108 because the loop on line 107 never started1a

108 households = session.query(Household).filter(Household.group_id == group.id).all() 

109 foods = ( 

110 session.query(IngredientFoodModel) 

111 .filter( 

112 IngredientFoodModel.group_id == group.id, 

113 IngredientFoodModel.on_hand == True, # noqa E712 

114 ) 

115 .all() 

116 ) 

117 

118 for food in foods: 

119 for household in households: 

120 session.execute( 

121 sa.text( 

122 "INSERT INTO households_to_ingredient_foods (household_id, food_id)" 

123 "VALUES (:household_id, :food_id)" 

124 ), 

125 { 

126 "household_id": GUID.convert_value_to_guid(household.id, dialect), 

127 "food_id": GUID.convert_value_to_guid(food.id, dialect), 

128 }, 

129 ) 

130 

131 

132def migrate_tools_on_hand_to_household(session: orm.Session): 1a

133 dialect = op.get_bind().dialect 1a

134 

135 for group in session.query(Group).all(): 135 ↛ 136line 135 didn't jump to line 136 because the loop on line 135 never started1a

136 households = session.query(Household).filter(Household.group_id == group.id).all() 

137 tools = ( 

138 session.query(Tool) 

139 .filter( 

140 Tool.group_id == group.id, 

141 Tool.on_hand == True, # noqa E712 

142 ) 

143 .all() 

144 ) 

145 

146 for tool in tools: 

147 for household in households: 

148 session.execute( 

149 sa.text("INSERT INTO households_to_tools (household_id, tool_id) VALUES (:household_id, :tool_id)"), 

150 { 

151 "household_id": GUID.convert_value_to_guid(household.id, dialect), 

152 "tool_id": GUID.convert_value_to_guid(tool.id, dialect), 

153 }, 

154 ) 

155 

156 

157def migrate_to_new_models(): 1a

158 bind = op.get_bind() 1a

159 session = orm.Session(bind=bind) 1a

160 

161 for migration_func in [ 1a

162 migrate_recipe_last_made_to_household, 

163 migrate_foods_on_hand_to_household, 

164 migrate_tools_on_hand_to_household, 

165 ]: 

166 try: 1a

167 logger.info(f"Running new model migration ({migration_func.__name__})") 1a

168 migration_func(session) 1a

169 session.commit() 1a

170 except Exception: 

171 session.rollback() 

172 logger.error(f"Error during new model migration ({migration_func.__name__})") 

173 raise 

174 

175 

176def upgrade(): 1a

177 # ### commands auto generated by Alembic - please adjust! ### 

178 op.create_table( 1a

179 "households_to_recipes", 

180 sa.Column("id", mealie.db.migration_types.GUID(), nullable=False), 

181 sa.Column("household_id", mealie.db.migration_types.GUID(), nullable=False), 

182 sa.Column("recipe_id", mealie.db.migration_types.GUID(), nullable=False), 

183 sa.Column("last_made", mealie.db.migration_types.NaiveDateTime(), nullable=True), 

184 sa.Column("created_at", mealie.db.migration_types.NaiveDateTime(), nullable=True), 

185 sa.Column("update_at", mealie.db.migration_types.NaiveDateTime(), nullable=True), 

186 sa.ForeignKeyConstraint( 

187 ["household_id"], 

188 ["households.id"], 

189 ), 

190 sa.ForeignKeyConstraint( 

191 ["recipe_id"], 

192 ["recipes.id"], 

193 ), 

194 sa.PrimaryKeyConstraint("id", "household_id", "recipe_id"), 

195 sa.UniqueConstraint("household_id", "recipe_id", name="household_id_recipe_id_key"), 

196 ) 

197 with op.batch_alter_table("households_to_recipes", schema=None) as batch_op: 1a

198 batch_op.create_index(batch_op.f("ix_households_to_recipes_created_at"), ["created_at"], unique=False) 1a

199 batch_op.create_index(batch_op.f("ix_households_to_recipes_household_id"), ["household_id"], unique=False) 1a

200 batch_op.create_index(batch_op.f("ix_households_to_recipes_recipe_id"), ["recipe_id"], unique=False) 1a

201 

202 op.create_table( 1a

203 "households_to_tools", 

204 sa.Column("household_id", mealie.db.migration_types.GUID(), nullable=True), 

205 sa.Column("tool_id", mealie.db.migration_types.GUID(), nullable=True), 

206 sa.ForeignKeyConstraint( 

207 ["household_id"], 

208 ["households.id"], 

209 ), 

210 sa.ForeignKeyConstraint( 

211 ["tool_id"], 

212 ["tools.id"], 

213 ), 

214 sa.UniqueConstraint("household_id", "tool_id", name="household_id_tool_id_key"), 

215 ) 

216 with op.batch_alter_table("households_to_tools", schema=None) as batch_op: 1a

217 batch_op.create_index(batch_op.f("ix_households_to_tools_household_id"), ["household_id"], unique=False) 1a

218 batch_op.create_index(batch_op.f("ix_households_to_tools_tool_id"), ["tool_id"], unique=False) 1a

219 

220 op.create_table( 1a

221 "households_to_ingredient_foods", 

222 sa.Column("household_id", mealie.db.migration_types.GUID(), nullable=True), 

223 sa.Column("food_id", mealie.db.migration_types.GUID(), nullable=True), 

224 sa.ForeignKeyConstraint( 

225 ["food_id"], 

226 ["ingredient_foods.id"], 

227 ), 

228 sa.ForeignKeyConstraint( 

229 ["household_id"], 

230 ["households.id"], 

231 ), 

232 sa.UniqueConstraint("household_id", "food_id", name="household_id_food_id_key"), 

233 ) 

234 with op.batch_alter_table("households_to_ingredient_foods", schema=None) as batch_op: 1a

235 batch_op.create_index(batch_op.f("ix_households_to_ingredient_foods_food_id"), ["food_id"], unique=False) 1a

236 batch_op.create_index( 1a

237 batch_op.f("ix_households_to_ingredient_foods_household_id"), ["household_id"], unique=False 

238 ) 

239 

240 # ### end Alembic commands ### 

241 

242 migrate_to_new_models() 1a

243 

244 

245def downgrade(): 1a

246 # ### commands auto generated by Alembic - please adjust! ### 

247 with op.batch_alter_table("households_to_ingredient_foods", schema=None) as batch_op: 

248 batch_op.drop_index(batch_op.f("ix_households_to_ingredient_foods_household_id")) 

249 batch_op.drop_index(batch_op.f("ix_households_to_ingredient_foods_food_id")) 

250 

251 op.drop_table("households_to_ingredient_foods") 

252 with op.batch_alter_table("households_to_tools", schema=None) as batch_op: 

253 batch_op.drop_index(batch_op.f("ix_households_to_tools_tool_id")) 

254 batch_op.drop_index(batch_op.f("ix_households_to_tools_household_id")) 

255 

256 op.drop_table("households_to_tools") 

257 with op.batch_alter_table("households_to_recipes", schema=None) as batch_op: 

258 batch_op.drop_index(batch_op.f("ix_households_to_recipes_recipe_id")) 

259 batch_op.drop_index(batch_op.f("ix_households_to_recipes_household_id")) 

260 batch_op.drop_index(batch_op.f("ix_households_to_recipes_created_at")) 

261 

262 op.drop_table("households_to_recipes") 

263 # ### end Alembic commands ###