Coverage for opt/mealie/lib/python3.12/site-packages/mealie/db/models/recipe/ingredient.py: 87%

180 statements  

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

1from typing import TYPE_CHECKING 1a

2 

3import sqlalchemy as sa 1a

4from pydantic import ConfigDict 1a

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

6from sqlalchemy.orm import Mapped, mapped_column 1a

7from sqlalchemy.orm.session import Session 1a

8 

9from mealie.db.models._model_base import BaseMixins, SqlAlchemyBase 1a

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

11from mealie.db.models.recipe.api_extras import IngredientFoodExtras, api_extras 1a

12 

13from .._model_utils.auto_init import auto_init 1a

14from .._model_utils.guid import GUID 1a

15 

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

17 from ..group import Group 

18 from ..household import Household 

19 

20 

21households_to_ingredient_foods = sa.Table( 1a

22 "households_to_ingredient_foods", 

23 SqlAlchemyBase.metadata, 

24 sa.Column("household_id", GUID, sa.ForeignKey("households.id"), index=True), 

25 sa.Column("food_id", GUID, sa.ForeignKey("ingredient_foods.id"), index=True), 

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

27) 

28 

29 

30class IngredientUnitModel(SqlAlchemyBase, BaseMixins): 1a

31 __tablename__ = "ingredient_units" 1a

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

33 

34 # ID Relationships 

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

36 group: Mapped["Group"] = orm.relationship("Group", back_populates="ingredient_units", foreign_keys=[group_id]) 1a

37 

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

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

40 description: Mapped[str | None] = mapped_column(String) 1a

41 abbreviation: Mapped[str | None] = mapped_column(String) 1a

42 plural_abbreviation: Mapped[str | None] = mapped_column(String) 1a

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

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

45 

46 ingredients: Mapped[list["RecipeIngredientModel"]] = orm.relationship( 1a

47 "RecipeIngredientModel", back_populates="unit" 

48 ) 

49 aliases: Mapped[list["IngredientUnitAliasModel"]] = orm.relationship( 1a

50 "IngredientUnitAliasModel", 

51 back_populates="unit", 

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

53 ) 

54 

55 # Automatically updated by sqlalchemy event, do not write to this manually 

56 name_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a

57 plural_name_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a

58 abbreviation_normalized: Mapped[str | None] = mapped_column(String, index=True) 1a

59 plural_abbreviation_normalized: Mapped[str | None] = mapped_column(String, index=True) 1a

60 

61 @auto_init() 1a

62 def __init__( 1a

63 self, 

64 session: Session, 

65 name: str | None = None, 

66 plural_name: str | None = None, 

67 abbreviation: str | None = None, 

68 plural_abbreviation: str | None = None, 

69 **_, 

70 ) -> None: 

71 if name is not None: 71 ↛ 73line 71 didn't jump to line 73 because the condition on line 71 was always true1kfgenodlhjmicb

72 self.name_normalized = self.normalize(name) 1kfgenodlhjmicb

73 if plural_name is not None: 1kfgenodlhjmicb

74 self.plural_name_normalized = self.normalize(plural_name) 1geodcb

75 if abbreviation is not None: 75 ↛ 77line 75 didn't jump to line 77 because the condition on line 75 was always true1kfgenodlhjmicb

76 self.abbreviation_normalized = self.normalize(abbreviation) 1kfgenodlhjmicb

77 if plural_abbreviation is not None: 1kfgenodlhjmicb

78 self.plural_abbreviation_normalized = self.normalize(plural_abbreviation) 1kfgenodlhjmicb

79 

80 tableargs = [ 1kfgenodlhjmicb

81 sa.UniqueConstraint("name", "group_id", name="ingredient_units_name_group_id_key"), 

82 sa.Index( 

83 "ix_ingredient_units_name_normalized", 

84 "name_normalized", 

85 unique=False, 

86 ), 

87 sa.Index( 

88 "ix_ingredient_units_plural_name_normalized", 

89 "plural_name_normalized", 

90 unique=False, 

91 ), 

92 sa.Index( 

93 "ix_ingredient_units_abbreviation_normalized", 

94 "abbreviation_normalized", 

95 unique=False, 

96 ), 

97 sa.Index( 

98 "ix_ingredient_units_plural_abbreviation_normalized", 

99 "plural_abbreviation_normalized", 

100 unique=False, 

101 ), 

102 ] 

103 

104 if session.get_bind().name == "postgresql": 104 ↛ 105line 104 didn't jump to line 105 because the condition on line 104 was never true1kfgenodlhjmicb

105 tableargs.extend( 

106 [ 

107 sa.Index( 

108 "ix_ingredient_units_name_normalized_gin", 

109 "name_normalized", 

110 unique=False, 

111 postgresql_using="gin", 

112 postgresql_ops={ 

113 "name_normalized": "gin_trgm_ops", 

114 }, 

115 ), 

116 sa.Index( 

117 "ix_ingredient_units_plural_name_normalized_gin", 

118 "name_normalized", 

119 unique=False, 

120 postgresql_using="gin", 

121 postgresql_ops={ 

122 "plural_name_normalized": "gin_trgm_ops", 

123 }, 

124 ), 

125 sa.Index( 

126 "ix_ingredient_units_abbreviation_normalized_gin", 

127 "abbreviation_normalized", 

128 unique=False, 

129 postgresql_using="gin", 

130 postgresql_ops={ 

131 "abbreviation_normalized": "gin_trgm_ops", 

132 }, 

133 ), 

134 sa.Index( 

135 "ix_ingredient_units_plural_abbreviation_normalized_gin", 

136 "plural_abbreviation_normalized", 

137 unique=False, 

138 postgresql_using="gin", 

139 postgresql_ops={ 

140 "plural_abbreviation_normalized": "gin_trgm_ops", 

141 }, 

142 ), 

143 ] 

144 ) 

145 

146 self.__table_args__ = tuple(tableargs) 1kfgenodlhjmicb

147 

148 

149class IngredientFoodModel(SqlAlchemyBase, BaseMixins): 1a

150 __tablename__ = "ingredient_foods" 1a

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

152 

153 # ID Relationships 

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

155 group: Mapped["Group"] = orm.relationship("Group", back_populates="ingredient_foods", foreign_keys=[group_id]) 1a

156 households_with_ingredient_food: Mapped[list["Household"]] = orm.relationship( 1a

157 "Household", secondary=households_to_ingredient_foods, back_populates="ingredient_foods_on_hand" 

158 ) 

159 

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

161 plural_name: Mapped[str | None] = mapped_column(String) 1a

162 description: Mapped[str | None] = mapped_column(String) 1a

163 

164 ingredients: Mapped[list["RecipeIngredientModel"]] = orm.relationship( 1a

165 "RecipeIngredientModel", back_populates="food" 

166 ) 

167 aliases: Mapped[list["IngredientFoodAliasModel"]] = orm.relationship( 1a

168 "IngredientFoodAliasModel", 

169 back_populates="food", 

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

171 ) 

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

173 

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

175 label: Mapped[MultiPurposeLabel | None] = orm.relationship(MultiPurposeLabel, uselist=False, back_populates="foods") 1a

176 

177 # Automatically updated by sqlalchemy event, do not write to this manually 

178 name_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a

179 plural_name_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a

180 

181 model_config = ConfigDict( 1a

182 exclude={ 

183 "households_with_ingredient_food", 

184 } 

185 ) 

186 

187 # Deprecated 

188 on_hand: Mapped[bool] = mapped_column(Boolean, default=False) 1a

189 

190 @api_extras 1a

191 @auto_init() 1a

192 def __init__( 1a

193 self, 

194 session: Session, 

195 group_id: GUID, 

196 name: str | None = None, 

197 plural_name: str | None = None, 

198 households_with_ingredient_food: list[str] | None = None, 

199 **_, 

200 ) -> None: 

201 from ..household import Household 1qkfgendlhjmicpb

202 

203 if name is not None: 203 ↛ 205line 203 didn't jump to line 205 because the condition on line 203 was always true1qkfgendlhjmicpb

204 self.name_normalized = self.normalize(name) 1qkfgendlhjmicpb

205 if plural_name is not None: 1qkfgendlhjmicpb

206 self.plural_name_normalized = self.normalize(plural_name) 1qfedhicpb

207 

208 if not households_with_ingredient_food: 1qkfgendlhjmicpb

209 self.households_with_ingredient_food = [] 1qkfgendlhjmicpb

210 else: 

211 self.households_with_ingredient_food = ( 1qfndhjicb

212 session.query(Household) 

213 .filter(Household.group_id == group_id, Household.slug.in_(households_with_ingredient_food)) 

214 .all() 

215 ) 

216 

217 tableargs = [ 1qkfgendlhjmicpb

218 sa.UniqueConstraint("name", "group_id", name="ingredient_foods_name_group_id_key"), 

219 sa.Index( 

220 "ix_ingredient_foods_name_normalized", 

221 "name_normalized", 

222 unique=False, 

223 ), 

224 sa.Index( 

225 "ix_ingredient_foods_plural_name_normalized", 

226 "plural_name_normalized", 

227 unique=False, 

228 ), 

229 ] 

230 

231 if session.get_bind().name == "postgresql": 231 ↛ 232line 231 didn't jump to line 232 because the condition on line 231 was never true1qkfgendlhjmicpb

232 tableargs.extend( 

233 [ 

234 sa.Index( 

235 "ix_ingredient_foods_name_normalized_gin", 

236 "name_normalized", 

237 unique=False, 

238 postgresql_using="gin", 

239 postgresql_ops={ 

240 "name_normalized": "gin_trgm_ops", 

241 }, 

242 ), 

243 sa.Index( 

244 "ix_ingredient_foods_plural_name_normalized_gin", 

245 "plural_name_normalized", 

246 unique=False, 

247 postgresql_using="gin", 

248 postgresql_ops={ 

249 "plural_name_normalized": "gin_trgm_ops", 

250 }, 

251 ), 

252 ] 

253 ) 

254 

255 self.__table_args__ = tuple(tableargs) 1qkfgendlhjmicpb

256 

257 

258class IngredientUnitAliasModel(SqlAlchemyBase, BaseMixins): 1a

259 __tablename__ = "ingredient_units_aliases" 1a

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

261 

262 unit_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("ingredient_units.id"), primary_key=True) 1a

263 unit: Mapped["IngredientUnitModel"] = orm.relationship("IngredientUnitModel", back_populates="aliases") 1a

264 

265 name: Mapped[str] = mapped_column(String) 1a

266 

267 # Automatically updated by sqlalchemy event, do not write to this manually 

268 name_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a

269 

270 @auto_init() 1a

271 def __init__(self, session: Session, name: str, **_) -> None: 1a

272 self.name_normalized = self.normalize(name) 1kfgeodlcb

273 tableargs = [ 1kfgeodlcb

274 sa.Index( 

275 "ix_ingredient_units_aliases_name_normalized", 

276 "name_normalized", 

277 unique=False, 

278 ), 

279 ] 

280 

281 if session.get_bind().name == "postgresql": 281 ↛ 282line 281 didn't jump to line 282 because the condition on line 281 was never true1kfgeodlcb

282 tableargs.extend( 

283 [ 

284 sa.Index( 

285 "ix_ingredient_units_aliases_name_normalized_gin", 

286 "name_normalized", 

287 unique=False, 

288 postgresql_using="gin", 

289 postgresql_ops={ 

290 "name_normalized": "gin_trgm_ops", 

291 }, 

292 ), 

293 ] 

294 ) 

295 

296 self.__table_args__ = tableargs 1kfgeodlcb

297 

298 

299class IngredientFoodAliasModel(SqlAlchemyBase, BaseMixins): 1a

300 __tablename__ = "ingredient_foods_aliases" 1a

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

302 

303 food_id: Mapped[GUID] = mapped_column(GUID, ForeignKey("ingredient_foods.id"), primary_key=True) 1a

304 food: Mapped["IngredientFoodModel"] = orm.relationship("IngredientFoodModel", back_populates="aliases") 1a

305 

306 name: Mapped[str] = mapped_column(String) 1a

307 

308 # Automatically updated by sqlalchemy event, do not write to this manually 

309 name_normalized: Mapped[str | None] = mapped_column(sa.String, index=True) 1a

310 

311 @auto_init() 1a

312 def __init__(self, session: Session, name: str, **_) -> None: 1a

313 self.name_normalized = self.normalize(name) 1fgedhjmicpb

314 tableargs = [ 1fgedhjmicpb

315 sa.Index( 

316 "ix_ingredient_foods_aliases_name_normalized", 

317 "name_normalized", 

318 unique=False, 

319 ), 

320 ] 

321 

322 if session.get_bind().name == "postgresql": 322 ↛ 323line 322 didn't jump to line 323 because the condition on line 322 was never true1fgedhjmicpb

323 tableargs.extend( 

324 [ 

325 sa.Index( 

326 "ix_ingredient_foods_aliases_name_normalized_gin", 

327 "name_normalized", 

328 unique=False, 

329 postgresql_using="gin", 

330 postgresql_ops={ 

331 "name_normalized": "gin_trgm_ops", 

332 }, 

333 ), 

334 ] 

335 ) 

336 

337 self.__table_args__ = tableargs 1fgedhjmicpb

338 

339 

340class RecipeIngredientModel(SqlAlchemyBase, BaseMixins): 1a

341 __tablename__ = "recipes_ingredients" 1a

342 id: Mapped[int] = mapped_column(Integer, primary_key=True) 1a

343 position: Mapped[int | None] = mapped_column(Integer, index=True) 1a

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

345 

346 title: Mapped[str | None] = mapped_column(String) # Section Header - Shows if Present 1a

347 note: Mapped[str | None] = mapped_column(String) # Force Show Text - Overrides Concat 1a

348 

349 # Scaling Items 

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

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

352 

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

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

355 quantity: Mapped[float | None] = mapped_column(Float) 1a

356 

357 original_text: Mapped[str | None] = mapped_column(String) 1a

358 

359 reference_id: Mapped[GUID | None] = mapped_column(GUID) # Reference Links 1a

360 

361 # Automatically updated by sqlalchemy event, do not write to this manually 

362 note_normalized: Mapped[str | None] = mapped_column(String, index=True) 1a

363 original_text_normalized: Mapped[str | None] = mapped_column(String, index=True) 1a

364 

365 @auto_init() 1a

366 def __init__( 1a

367 self, 

368 session: Session, 

369 note: str | None = None, 

370 orginal_text: str | None = None, 

371 **_, 

372 ) -> None: 

373 # SQLAlchemy events do not seem to register things that are set during auto_init 

374 if note is not None: 374 ↛ 377line 374 didn't jump to line 377 because the condition on line 374 was always true1rb

375 self.note_normalized = self.normalize(note) 1rb

376 

377 if orginal_text is not None: 377 ↛ 378line 377 didn't jump to line 378 because the condition on line 377 was never true1rb

378 self.orginal_text = self.normalize(orginal_text) 

379 

380 tableargs = [ # base set of indices 1rb

381 sa.Index( 

382 "ix_recipes_ingredients_note_normalized", 

383 "note_normalized", 

384 unique=False, 

385 ), 

386 sa.Index( 

387 "ix_recipes_ingredients_original_text_normalized", 

388 "original_text_normalized", 

389 unique=False, 

390 ), 

391 ] 

392 if session.get_bind().name == "postgresql": 392 ↛ 393line 392 didn't jump to line 393 because the condition on line 392 was never true1rb

393 tableargs.extend( 

394 [ 

395 sa.Index( 

396 "ix_recipes_ingredients_note_normalized_gin", 

397 "note_normalized", 

398 unique=False, 

399 postgresql_using="gin", 

400 postgresql_ops={ 

401 "note_normalized": "gin_trgm_ops", 

402 }, 

403 ), 

404 sa.Index( 

405 "ix_recipes_ingredients_original_text_normalized_gin", 

406 "original_text", 

407 unique=False, 

408 postgresql_using="gin", 

409 postgresql_ops={ 

410 "original_text_normalized": "gin_trgm_ops", 

411 }, 

412 ), 

413 ] 

414 ) 

415 # add indices 

416 self.__table_args__ = tuple(tableargs) 1rb

417 

418 

419@event.listens_for(IngredientUnitModel.name, "set") 1a

420def receive_unit_name(target: IngredientUnitModel, value: str | None, oldvalue, initiator): 1a

421 if value is not None: 421 ↛ 424line 421 didn't jump to line 424 because the condition on line 421 was always true1kfgenodlhjmicb

422 target.name_normalized = IngredientUnitModel.normalize(value) 1kfgenodlhjmicb

423 else: 

424 target.name_normalized = None 

425 

426 

427@event.listens_for(IngredientUnitModel.plural_name, "set") 1a

428def receive_plural_unit_name(target: IngredientUnitModel, value: str | None, oldvalue, initiator): 1a

429 if value is not None: 1kfgenodlhjmicb

430 target.plural_name_normalized = IngredientUnitModel.normalize(value) 1geodcb

431 else: 

432 target.plural_name_normalized = None 1kfgenodlhjmicb

433 

434 

435@event.listens_for(IngredientUnitModel.abbreviation, "set") 1a

436def receive_unit_abbreviation(target: IngredientUnitModel, value: str | None, oldvalue, initiator): 1a

437 if value is not None: 437 ↛ 440line 437 didn't jump to line 440 because the condition on line 437 was always true1kfgenodlhjmicb

438 target.abbreviation_normalized = IngredientUnitModel.normalize(value) 1kfgenodlhjmicb

439 else: 

440 target.abbreviation_normalized = None 

441 

442 

443@event.listens_for(IngredientUnitModel.plural_abbreviation, "set") 1a

444def receive_unit_plural_abbreviation(target: IngredientUnitModel, value: str | None, oldvalue, initiator): 1a

445 if value is not None: 1kfgenodlhjmicb

446 target.plural_abbreviation_normalized = IngredientUnitModel.normalize(value) 1kfgenodlhjmicb

447 else: 

448 target.plural_abbreviation_normalized = None 1gnohicb

449 

450 

451@event.listens_for(IngredientFoodModel.name, "set") 1a

452def receive_food_name(target: IngredientFoodModel, value: str | None, oldvalue, initiator): 1a

453 if value is not None: 453 ↛ 456line 453 didn't jump to line 456 because the condition on line 453 was always true1qkfgendlhjmicpb

454 target.name_normalized = IngredientFoodModel.normalize(value) 1qkfgendlhjmicpb

455 else: 

456 target.name_normalized = None 

457 

458 

459@event.listens_for(IngredientFoodModel.plural_name, "set") 1a

460def receive_food_plural_name(target: IngredientFoodModel, value: str | None, oldvalue, initiator): 1a

461 if value is not None: 1qkfgendlhjmicpb

462 target.plural_name_normalized = IngredientFoodModel.normalize(value) 1qfedhicpb

463 else: 

464 target.plural_name_normalized = None 1qkfgendlhjmicb

465 

466 

467@event.listens_for(IngredientUnitAliasModel.name, "set") 1a

468def receive_unit_alias_name(target: IngredientUnitAliasModel, value: str, oldvalue, initiator): 1a

469 target.name_normalized = IngredientUnitAliasModel.normalize(value) 1kfgeodlcb

470 

471 

472@event.listens_for(IngredientFoodAliasModel.name, "set") 1a

473def receive_food_alias_name(target: IngredientFoodAliasModel, value: str, oldvalue, initiator): 1a

474 target.name_normalized = IngredientFoodAliasModel.normalize(value) 1fgedhjmicpb

475 

476 

477@event.listens_for(RecipeIngredientModel.note, "set") 1a

478def receive_ingredient_note(target: RecipeIngredientModel, value: str | None, oldvalue, initiator): 1a

479 if value is not None: 479 ↛ 482line 479 didn't jump to line 482 because the condition on line 479 was always true1rb

480 target.note_normalized = RecipeIngredientModel.normalize(value) 1rb

481 else: 

482 target.note_normalized = None 

483 

484 

485@event.listens_for(RecipeIngredientModel.original_text, "set") 1a

486def receive_ingredient_original_text(target: RecipeIngredientModel, value: str | None, oldvalue, initiator): 1a

487 if value is not None: 487 ↛ 488line 487 didn't jump to line 488 because the condition on line 487 was never true1rb

488 target.original_text_normalized = RecipeIngredientModel.normalize(value) 

489 else: 

490 target.original_text_normalized = None 1rb