Coverage for opt/mealie/lib/python3.12/site-packages/mealie/alembic/versions/2023-10-04-14.29.26_dded3119c1fe_added_unique_constraints.py: 61%

1"""added unique constraints 

2 

3Revision ID: dded3119c1fe 

4Revises: 0341b154f79a 

5Create Date: 2023-10-04 14:29:26.688065 

6 

7""" 

8 

9from collections import defaultdict 

10from dataclasses import dataclass 

11from typing import Any 

12 

13import sqlalchemy as sa 

14from pydantic import UUID4 

15from sqlalchemy import orm 

16from sqlalchemy.orm import Session, load_only 

17 

18from alembic import op 

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

20from mealie.db.models.labels import MultiPurposeLabel 

21from mealie.db.models.recipe.ingredient import IngredientFoodModel, IngredientUnitModel, RecipeIngredientModel 

22 

23# revision identifiers, used by Alembic. 

24revision = "dded3119c1fe" 

25down_revision = "0341b154f79a" 

26branch_labels: str | tuple[str, ...] | None = None 

27depends_on: str | tuple[str, ...] | None = None 

28 

29 

30# Intermediate table definitions 

31class SqlAlchemyBase(orm.DeclarativeBase): 

32 pass 

33 

34 

35class ShoppingList(SqlAlchemyBase): 

36 __tablename__ = "shopping_lists" 

37 

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

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

40 

41 

42class ShoppingListItem(SqlAlchemyBase): 

43 __tablename__ = "shopping_list_items" 

44 

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

46 food_id: orm.Mapped[GUID] = orm.mapped_column(GUID, sa.ForeignKey("ingredient_foods.id")) 

47 unit_id: orm.Mapped[GUID] = orm.mapped_column(GUID, sa.ForeignKey("ingredient_units.id")) 

48 label_id: orm.Mapped[GUID] = orm.mapped_column(GUID, sa.ForeignKey("multi_purpose_labels.id")) 

49 

50 

51@dataclass 

52class TableMeta: 

53 tablename: str 

54 pk_1: str 

55 pk_2: str 

56 

57 @classmethod 

58 def composite_pk(self, pk_1_val: Any, pk_2_val: Any) -> str: 

59 return "$$".join([pk_1_val, pk_2_val]) 

60 

61 

62def _is_postgres(): 

63 return op.get_context().dialect.name == "postgresql" 

64 

65 

66def _get_duplicates(session: Session, model: orm.DeclarativeBase) -> defaultdict[str, list]: 

67 duplicate_map: defaultdict[str, list] = defaultdict(list) 

68 

69 query = session.execute(sa.text(f"SELECT id, group_id, name FROM {model.__tablename__}")) 

70 for row in query.all(): 70 ↛ 71line 70 didn't jump to line 71 because the loop on line 70 never started

71 id, group_id, name = row 

72 key = f"{group_id}$${name}" 

73 duplicate_map[key].append(id) 

74 

75 return duplicate_map 

76 

77 

78def _resolve_duplicate_food( 

79 session: Session, 

80 keep_food_id: UUID4, 

81 dupe_food_id: UUID4, 

82): 

83 for shopping_list_item in ( 

84 session.query(ShoppingListItem) 

85 .options(load_only(ShoppingListItem.id, ShoppingListItem.food_id)) 

86 .filter_by(food_id=dupe_food_id) 

87 .all() 

88 ): 

89 shopping_list_item.food_id = keep_food_id 

90 

91 for recipe_ingredient in ( 

92 session.query(RecipeIngredientModel) 

93 .options(load_only(RecipeIngredientModel.id, RecipeIngredientModel.food_id)) 

94 .filter_by(food_id=dupe_food_id) 

95 .all() 

96 ): 

97 recipe_ingredient.food_id = keep_food_id 

98 

99 session.commit() 

100 session.execute( 

101 sa.text(f"DELETE FROM {IngredientFoodModel.__tablename__} WHERE id=:id").bindparams(id=dupe_food_id) 

102 ) 

103 session.commit() 

104 

105 

106def _resolve_duplicate_unit( 

107 session: Session, 

108 keep_unit_id: UUID4, 

109 dupe_unit_id: UUID4, 

110): 

111 for shopping_list_item in ( 

112 session.query(ShoppingListItem) 

113 .options(load_only(ShoppingListItem.id, ShoppingListItem.unit_id)) 

114 .filter_by(unit_id=dupe_unit_id) 

115 .all() 

116 ): 

117 shopping_list_item.unit_id = keep_unit_id 

118 

119 for recipe_ingredient in ( 

120 session.query(RecipeIngredientModel) 

121 .options(load_only(RecipeIngredientModel.id, RecipeIngredientModel.unit_id)) 

122 .filter_by(unit_id=dupe_unit_id) 

123 .all() 

124 ): 

125 recipe_ingredient.unit_id = keep_unit_id 

126 

127 session.commit() 

128 session.execute( 

129 sa.text(f"DELETE FROM {IngredientUnitModel.__tablename__} WHERE id=:id").bindparams(id=dupe_unit_id) 

130 ) 

131 session.commit() 

132 

133 

134def _resolve_duplicate_label( 

135 session: Session, 

136 keep_label_id: UUID4, 

137 dupe_label_id: UUID4, 

138): 

139 for shopping_list_item in ( 

140 session.query(ShoppingListItem) 

141 .options(load_only(ShoppingListItem.id, ShoppingListItem.label_id)) 

142 .filter_by(label_id=dupe_label_id) 

143 .all() 

144 ): 

145 shopping_list_item.label_id = keep_label_id 

146 

147 for ingredient_food in ( 

148 session.query(IngredientFoodModel) 

149 .options(load_only(IngredientFoodModel.id, IngredientFoodModel.label_id)) 

150 .filter_by(label_id=dupe_label_id) 

151 .all() 

152 ): 

153 ingredient_food.label_id = keep_label_id 

154 

155 session.commit() 

156 session.execute(sa.text(f"DELETE FROM {MultiPurposeLabel.__tablename__} WHERE id=:id").bindparams(id=dupe_label_id)) 

157 session.commit() 

158 

159 

160def _resolve_duplicate_foods_units_labels(session: Session): 

161 for model, resolve_func in [ 

162 (IngredientFoodModel, _resolve_duplicate_food), 

163 (IngredientUnitModel, _resolve_duplicate_unit), 

164 (MultiPurposeLabel, _resolve_duplicate_label), 

165 ]: 

166 duplicate_map = _get_duplicates(session, model) 

167 for ids in duplicate_map.values(): 167 ↛ 168line 167 didn't jump to line 168 because the loop on line 167 never started

168 if len(ids) < 2: 

169 continue 

170 

171 keep_id = ids[0] 

172 for dupe_id in ids[1:]: 

173 resolve_func(session, keep_id, dupe_id) 

174 

175 

176def _remove_duplicates_from_m2m_table(session: Session, table_meta: TableMeta): 

177 if _is_postgres(): 177 ↛ 178line 177 didn't jump to line 178 because the condition on line 177 was never true

178 default_pk = "CTID" 

179 else: 

180 default_pk = "ROWID" 

181 

182 # some of these tables are missing defined unique pks, so we have to rely on the database default pk 

183 query = sa.text( 

184 f""" 

185 DELETE FROM {table_meta.tablename} 

186 WHERE EXISTS ( 

187 SELECT 1 FROM {table_meta.tablename} t2 

188 WHERE {table_meta.tablename}.{table_meta.pk_1} = t2.{table_meta.pk_1} 

189 AND {table_meta.tablename}.{table_meta.pk_2} = t2.{table_meta.pk_2} 

190 AND {table_meta.tablename}.{default_pk} > t2.{default_pk} 

191 ) 

192 """ 

193 ) 

194 

195 session.execute(query) 

196 session.commit() 

197 

198 

199def _remove_duplicates_from_m2m_tables(session: Session, table_metas: list[TableMeta]): 

200 for table_meta in table_metas: 

201 _remove_duplicates_from_m2m_table(session, table_meta) 

202 

203 

204def upgrade(): 

205 bind = op.get_bind() 

206 session = Session(bind=bind) 

207 

208 _resolve_duplicate_foods_units_labels(session) 

209 _remove_duplicates_from_m2m_tables( 

210 session, 

211 [ 

212 TableMeta("cookbooks_to_categories", "cookbook_id", "category_id"), 

213 TableMeta("cookbooks_to_tags", "cookbook_id", "tag_id"), 

214 TableMeta("cookbooks_to_tools", "cookbook_id", "tool_id"), 

215 TableMeta("group_to_categories", "group_id", "category_id"), 

216 TableMeta("plan_rules_to_categories", "group_plan_rule_id", "category_id"), 

217 TableMeta("plan_rules_to_tags", "plan_rule_id", "tag_id"), 

218 TableMeta("recipes_to_categories", "recipe_id", "category_id"), 

219 TableMeta("recipes_to_tags", "recipe_id", "tag_id"), 

220 TableMeta("recipes_to_tools", "recipe_id", "tool_id"), 

221 TableMeta("users_to_favorites", "user_id", "recipe_id"), 

222 TableMeta("shopping_lists_multi_purpose_labels", "shopping_list_id", "label_id"), 

223 ], 

224 ) 

225 

226 session.commit() 

227 

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

229 # we use batch_alter_table here because otherwise this fails on sqlite 

230 # M2M 

231 with op.batch_alter_table("cookbooks_to_categories") as batch_op: 

232 batch_op.create_unique_constraint("cookbook_id_category_id_key", ["cookbook_id", "category_id"]) 

233 

234 with op.batch_alter_table("cookbooks_to_tags") as batch_op: 

235 batch_op.create_unique_constraint("cookbook_id_tag_id_key", ["cookbook_id", "tag_id"]) 

236 

237 with op.batch_alter_table("cookbooks_to_tools") as batch_op: 

238 batch_op.create_unique_constraint("cookbook_id_tool_id_key", ["cookbook_id", "tool_id"]) 

239 

240 with op.batch_alter_table("group_to_categories") as batch_op: 

241 batch_op.create_unique_constraint("group_id_category_id_key", ["group_id", "category_id"]) 

242 

243 with op.batch_alter_table("plan_rules_to_categories") as batch_op: 

244 batch_op.create_unique_constraint("group_plan_rule_id_category_id_key", ["group_plan_rule_id", "category_id"]) 

245 

246 with op.batch_alter_table("plan_rules_to_tags") as batch_op: 

247 batch_op.create_unique_constraint("plan_rule_id_tag_id_key", ["plan_rule_id", "tag_id"]) 

248 

249 with op.batch_alter_table("recipes_to_categories") as batch_op: 

250 batch_op.create_unique_constraint("recipe_id_category_id_key", ["recipe_id", "category_id"]) 

251 

252 with op.batch_alter_table("recipes_to_tags") as batch_op: 

253 batch_op.create_unique_constraint("recipe_id_tag_id_key", ["recipe_id", "tag_id"]) 

254 

255 with op.batch_alter_table("recipes_to_tools") as batch_op: 

256 batch_op.create_unique_constraint("recipe_id_tool_id_key", ["recipe_id", "tool_id"]) 

257 

258 with op.batch_alter_table("users_to_favorites") as batch_op: 

259 batch_op.create_unique_constraint("user_id_recipe_id_key", ["user_id", "recipe_id"]) 

260 

261 with op.batch_alter_table("shopping_lists_multi_purpose_labels") as batch_op: 

262 batch_op.create_unique_constraint("shopping_list_id_label_id_key", ["shopping_list_id", "label_id"]) 

263 

264 # Foods/Units/Labels 

265 with op.batch_alter_table("ingredient_foods") as batch_op: 

266 batch_op.create_unique_constraint("ingredient_foods_name_group_id_key", ["name", "group_id"]) 

267 

268 with op.batch_alter_table("ingredient_units") as batch_op: 

269 batch_op.create_unique_constraint("ingredient_units_name_group_id_key", ["name", "group_id"]) 

270 

271 with op.batch_alter_table("multi_purpose_labels") as batch_op: 

272 batch_op.create_unique_constraint("multi_purpose_labels_name_group_id_key", ["name", "group_id"]) 

273 

274 op.create_index( 

275 op.f("ix_shopping_lists_multi_purpose_labels_created_at"), 

276 "shopping_lists_multi_purpose_labels", 

277 ["created_at"], 

278 unique=False, 

279 ) 

280 # ### end Alembic commands ### 

281 

282 

283def downgrade(): 

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

285 

286 # M2M 

287 op.drop_constraint("user_id_recipe_id_key", "users_to_favorites", type_="unique") 

288 op.drop_index( 

289 op.f("ix_shopping_lists_multi_purpose_labels_created_at"), table_name="shopping_lists_multi_purpose_labels" 

290 ) 

291 op.drop_constraint("recipe_id_tool_id_key", "recipes_to_tools", type_="unique") 

292 op.drop_constraint("recipe_id_tag_id_key", "recipes_to_tags", type_="unique") 

293 op.drop_constraint("recipe_id_category_id_key", "recipes_to_categories", type_="unique") 

294 op.drop_constraint("plan_rule_id_tag_id_key", "plan_rules_to_tags", type_="unique") 

295 op.drop_constraint("group_plan_rule_id_category_id_key", "plan_rules_to_categories", type_="unique") 

296 op.drop_constraint("group_id_category_id_key", "group_to_categories", type_="unique") 

297 op.drop_constraint("cookbook_id_tool_id_key", "cookbooks_to_tools", type_="unique") 

298 op.drop_constraint("cookbook_id_tag_id_key", "cookbooks_to_tags", type_="unique") 

299 op.drop_constraint("cookbook_id_category_id_key", "cookbooks_to_categories", type_="unique") 

300 op.drop_constraint("shopping_list_id_label_id_key", "shopping_lists_multi_purpose_labels", type_="unique") 

301 

302 # Foods/Units/Labels 

303 op.drop_constraint("multi_purpose_labels_name_group_id_key", "multi_purpose_labels", type_="unique") 

304 op.drop_constraint("ingredient_units_name_group_id_key", "ingredient_units", type_="unique") 

305 op.drop_constraint("ingredient_foods_name_group_id_key", "ingredient_foods", type_="unique") 

306 # ### end Alembic commands ###