mirror of
https://github.com/mealie-recipes/mealie.git
synced 2026-01-22 00:41:19 -05:00
Refactor Shopping List API (#2021)
* tidied up shopping list item models redefined recipe refs and updated models added calculated display attribute to unify shopping list item rendering added validation to use a food's label if an item's label is null * fixed schema reference * refactored shopping list item service route all operations through one central method to account for edgecases return item collections for all operations to account for merging consolidate recipe items before sending them to the shopping list * made fractions prettier * replaced redundant display text util * fixed edgecase for zero quantity items on a recipe * fix for pre-merging recipe ingredients * fixed edgecase for merging create_items together * fixed bug with merged updated items creating dupes * added test for self-removing recipe ref * update items are now merged w/ existing items * refactored service to make it easier to read * added a lot of tests * made it so checked items are never merged * fixed bug with dragging + re-ordering * fix for postgres cascade issue * added prevalidator to recipe ref to avoid db error
This commit is contained in:
Michael Genson
GitHub
@@ -6,12 +6,16 @@ from mealie.core.exceptions import UnexpectedNone
|
||||
from mealie.repos.repository_factory import AllRepositories
|
||||
from mealie.schema.group import ShoppingListItemCreate, ShoppingListOut
|
||||
from mealie.schema.group.group_shopping_list import (
|
||||
ShoppingListItemBase,
|
||||
ShoppingListItemOut,
|
||||
ShoppingListItemRecipeRef,
|
||||
ShoppingListItemRecipeRefCreate,
|
||||
ShoppingListItemRecipeRefOut,
|
||||
ShoppingListItemsCollectionOut,
|
||||
ShoppingListItemUpdate,
|
||||
ShoppingListItemUpdateBulk,
|
||||
)
|
||||
from mealie.schema.recipe import Recipe
|
||||
from mealie.schema.recipe.recipe_ingredient import IngredientFood, IngredientUnit
|
||||
from mealie.schema.response.pagination import PaginationQuery
|
||||
|
||||
|
||||
class ShoppingListService:
|
||||
@@ -23,240 +27,359 @@ class ShoppingListService:
|
||||
self.list_refs = repos.group_shopping_list_recipe_refs
|
||||
|
||||
@staticmethod
|
||||
def can_merge(item1: ShoppingListItemOut, item2: ShoppingListItemOut) -> bool:
|
||||
"""
|
||||
can_merge checks if the two items can be merged together.
|
||||
"""
|
||||
def can_merge(item1: ShoppingListItemBase, item2: ShoppingListItemBase) -> bool:
|
||||
"""Check to see if this item can be merged with another item"""
|
||||
|
||||
# Check if items are both checked or both unchecked
|
||||
if item1.checked != item2.checked:
|
||||
if any(
|
||||
[
|
||||
item1.checked,
|
||||
item2.checked,
|
||||
item1.food_id != item2.food_id,
|
||||
item1.unit_id != item2.unit_id,
|
||||
]
|
||||
):
|
||||
return False
|
||||
|
||||
# Check if foods are equal
|
||||
foods_is_none = item1.food_id is None and item2.food_id is None
|
||||
foods_not_none = not foods_is_none
|
||||
foods_equal = item1.food_id == item2.food_id
|
||||
# if foods match, we can merge, otherwise compare the notes
|
||||
return bool(item1.food_id) or item1.note == item2.note
|
||||
|
||||
# Check if units are equal
|
||||
units_is_none = item1.unit_id is None and item2.unit_id is None
|
||||
units_not_none = not units_is_none
|
||||
units_equal = item1.unit_id == item2.unit_id
|
||||
|
||||
# Check if notes are equal
|
||||
if foods_is_none and units_is_none:
|
||||
return item1.note == item2.note
|
||||
|
||||
if foods_not_none and units_not_none:
|
||||
return foods_equal and units_equal
|
||||
|
||||
if foods_not_none:
|
||||
return foods_equal
|
||||
|
||||
return False
|
||||
|
||||
def consolidate_list_items(self, item_list: list[ShoppingListItemOut]) -> list[ShoppingListItemOut]:
|
||||
@staticmethod
|
||||
def merge_items(
|
||||
from_item: ShoppingListItemCreate | ShoppingListItemUpdateBulk,
|
||||
to_item: ShoppingListItemCreate | ShoppingListItemUpdateBulk | ShoppingListItemOut,
|
||||
) -> ShoppingListItemUpdate:
|
||||
"""
|
||||
iterates through the shopping list provided and returns
|
||||
a consolidated list where all items that are matched against multiple values are
|
||||
de-duplicated and only the first item is kept where the quantity is updated accordingly.
|
||||
Takes an item and merges it into an already-existing item, then returns a copy
|
||||
|
||||
Attributes of the `to_item` take priority over the `from_item`, except extras with overlapping keys
|
||||
"""
|
||||
|
||||
consolidated_list: list[ShoppingListItemOut] = []
|
||||
checked_items: list[int] = []
|
||||
to_item.quantity += from_item.quantity
|
||||
if to_item.note != from_item.note:
|
||||
to_item.note = " | ".join([note for note in [to_item.note, from_item.note] if note])
|
||||
|
||||
for base_index, base_item in enumerate(item_list):
|
||||
if base_index in checked_items:
|
||||
if to_item.extras and from_item.extras:
|
||||
to_item.extras.update(from_item.extras)
|
||||
|
||||
updated_refs = {ref.recipe_id: ref for ref in from_item.recipe_references}
|
||||
for to_ref in to_item.recipe_references:
|
||||
if to_ref.recipe_id not in updated_refs:
|
||||
updated_refs[to_ref.recipe_id] = to_ref
|
||||
continue
|
||||
|
||||
checked_items.append(base_index)
|
||||
for inner_index, inner_item in enumerate(item_list):
|
||||
if inner_index in checked_items:
|
||||
# merge recipe scales
|
||||
base_ref = updated_refs[to_ref.recipe_id]
|
||||
|
||||
# if the scale is missing we assume it's 1 for backwards compatibility
|
||||
# if the scale is 0 we leave it alone
|
||||
if base_ref.recipe_scale is None:
|
||||
base_ref.recipe_scale = 1
|
||||
|
||||
if to_ref.recipe_scale is None:
|
||||
to_ref.recipe_scale = 1
|
||||
|
||||
base_ref.recipe_scale += to_ref.recipe_scale
|
||||
|
||||
return to_item.cast(ShoppingListItemUpdate, recipe_references=list(updated_refs.values()))
|
||||
|
||||
def remove_unused_recipe_references(self, shopping_list_id: UUID4) -> None:
|
||||
shopping_list = cast(ShoppingListOut, self.shopping_lists.get_one(shopping_list_id))
|
||||
|
||||
recipe_ids_to_keep: set[UUID4] = set()
|
||||
for item in shopping_list.list_items:
|
||||
recipe_ids_to_keep.update([ref.recipe_id for ref in item.recipe_references])
|
||||
|
||||
list_refs_to_delete: set[UUID4] = set()
|
||||
for list_ref in shopping_list.recipe_references:
|
||||
if list_ref.recipe_id not in recipe_ids_to_keep:
|
||||
list_refs_to_delete.add(list_ref.id)
|
||||
|
||||
if list_refs_to_delete:
|
||||
self.list_refs.delete_many(list_refs_to_delete)
|
||||
|
||||
def bulk_create_items(self, create_items: list[ShoppingListItemCreate]) -> ShoppingListItemsCollectionOut:
|
||||
# consolidate items to be created
|
||||
consolidated_create_items: list[ShoppingListItemCreate] = []
|
||||
for create_item in create_items:
|
||||
merged = False
|
||||
for filtered_item in consolidated_create_items:
|
||||
if not self.can_merge(create_item, filtered_item):
|
||||
continue
|
||||
|
||||
if ShoppingListService.can_merge(base_item, inner_item):
|
||||
# Set Quantity
|
||||
base_item.quantity += inner_item.quantity
|
||||
filtered_item = self.merge_items(create_item, filtered_item).cast(ShoppingListItemCreate)
|
||||
merged = True
|
||||
break
|
||||
|
||||
# Set References
|
||||
refs = {ref.recipe_id: ref for ref in base_item.recipe_references}
|
||||
for inner_ref in inner_item.recipe_references:
|
||||
if inner_ref.recipe_id not in refs:
|
||||
refs[inner_ref.recipe_id] = inner_ref
|
||||
if not merged:
|
||||
consolidated_create_items.append(create_item)
|
||||
|
||||
else:
|
||||
# merge recipe scales
|
||||
base_ref = refs[inner_ref.recipe_id]
|
||||
create_items = consolidated_create_items
|
||||
filtered_create_items: list[ShoppingListItemCreate] = []
|
||||
|
||||
# if the scale is missing we assume it's 1 for backwards compatibility
|
||||
# if the scale is 0 we leave it alone
|
||||
if base_ref.recipe_scale is None:
|
||||
base_ref.recipe_scale = 1
|
||||
# check to see if we can merge into any existing items
|
||||
update_items: list[ShoppingListItemUpdateBulk] = []
|
||||
existing_items_map: dict[UUID4, list[ShoppingListItemOut]] = {}
|
||||
for create_item in create_items:
|
||||
if create_item.shopping_list_id not in existing_items_map:
|
||||
query = PaginationQuery(
|
||||
per_page=-1, query_filter=f"shopping_list_id={create_item.shopping_list_id} AND checked=false"
|
||||
)
|
||||
items_data = self.list_items.page_all(query)
|
||||
existing_items_map[create_item.shopping_list_id] = items_data.items
|
||||
|
||||
if inner_ref.recipe_scale is None:
|
||||
inner_ref.recipe_scale = 1
|
||||
merged = False
|
||||
for existing_item in existing_items_map[create_item.shopping_list_id]:
|
||||
if not self.can_merge(existing_item, create_item):
|
||||
continue
|
||||
|
||||
base_ref.recipe_scale += inner_ref.recipe_scale
|
||||
updated_existing_item = self.merge_items(create_item, existing_item).cast(
|
||||
ShoppingListItemUpdateBulk, id=existing_item.id
|
||||
)
|
||||
update_items.append(updated_existing_item.cast(ShoppingListItemUpdateBulk, id=existing_item.id))
|
||||
merged = True
|
||||
break
|
||||
|
||||
base_item.recipe_references = list(refs.values())
|
||||
checked_items.append(inner_index)
|
||||
if merged or create_item.quantity < 0:
|
||||
continue
|
||||
|
||||
consolidated_list.append(base_item)
|
||||
# create the item
|
||||
if create_item.checked:
|
||||
# checked items should not have recipe references
|
||||
create_item.recipe_references = []
|
||||
|
||||
return consolidated_list
|
||||
filtered_create_items.append(create_item)
|
||||
|
||||
def consolidate_and_save(
|
||||
self, data: list[ShoppingListItemUpdate]
|
||||
) -> tuple[list[ShoppingListItemOut], list[ShoppingListItemOut]]:
|
||||
"""
|
||||
returns:
|
||||
- updated_shopping_list_items
|
||||
- deleted_shopping_list_items
|
||||
"""
|
||||
# TODO: Convert to update many with single call
|
||||
created_items = cast(
|
||||
list[ShoppingListItemOut],
|
||||
self.list_items.create_many(filtered_create_items) if filtered_create_items else [], # type: ignore
|
||||
)
|
||||
|
||||
all_updates = []
|
||||
all_deletes = []
|
||||
keep_ids = []
|
||||
updated_items = cast(
|
||||
list[ShoppingListItemOut],
|
||||
self.list_items.update_many(update_items) if update_items else [], # type: ignore
|
||||
)
|
||||
|
||||
for item in self.consolidate_list_items(data): # type: ignore
|
||||
updated_data = self.list_items.update(item.id, item)
|
||||
all_updates.append(updated_data)
|
||||
keep_ids.append(updated_data.id)
|
||||
for list_id in set(item.shopping_list_id for item in created_items + updated_items):
|
||||
self.remove_unused_recipe_references(list_id)
|
||||
|
||||
for item in data: # type: ignore
|
||||
if item.id not in keep_ids:
|
||||
self.list_items.delete(item.id)
|
||||
all_deletes.append(item)
|
||||
return ShoppingListItemsCollectionOut(
|
||||
created_items=created_items, updated_items=updated_items, deleted_items=[]
|
||||
)
|
||||
|
||||
return all_updates, all_deletes
|
||||
def bulk_update_items(self, update_items: list[ShoppingListItemUpdateBulk]) -> ShoppingListItemsCollectionOut:
|
||||
# consolidate items to be created
|
||||
consolidated_update_items: list[ShoppingListItemUpdateBulk] = []
|
||||
delete_items: set[UUID4] = set()
|
||||
seen_update_ids: set[UUID4] = set()
|
||||
for update_item in update_items:
|
||||
# if the same item appears multiple times in one request, ignore all but the first instance
|
||||
if update_item.id in seen_update_ids:
|
||||
continue
|
||||
|
||||
# =======================================================================
|
||||
# Methods
|
||||
seen_update_ids.add(update_item.id)
|
||||
|
||||
def add_recipe_ingredients_to_list(
|
||||
self, list_id: UUID4, recipe_id: UUID4, recipe_increment: float = 1
|
||||
) -> tuple[ShoppingListOut, list[ShoppingListItemOut], list[ShoppingListItemOut], list[ShoppingListItemOut]]:
|
||||
"""
|
||||
returns:
|
||||
- updated_shopping_list
|
||||
- new_shopping_list_items
|
||||
- updated_shopping_list_items
|
||||
- deleted_shopping_list_items
|
||||
"""
|
||||
recipe: Recipe | None = self.repos.recipes.get_one(recipe_id, "id")
|
||||
merged = False
|
||||
for filtered_item in consolidated_update_items:
|
||||
if not self.can_merge(update_item, filtered_item):
|
||||
continue
|
||||
|
||||
filtered_item = self.merge_items(update_item, filtered_item).cast(
|
||||
ShoppingListItemUpdateBulk, id=filtered_item.id
|
||||
)
|
||||
delete_items.add(update_item.id)
|
||||
merged = True
|
||||
break
|
||||
|
||||
if not merged:
|
||||
consolidated_update_items.append(update_item)
|
||||
|
||||
update_items = consolidated_update_items
|
||||
|
||||
# check to see if we can merge into any existing items
|
||||
filtered_update_items: list[ShoppingListItemUpdateBulk] = []
|
||||
existing_items_map: dict[UUID4, list[ShoppingListItemOut]] = {}
|
||||
for update_item in update_items:
|
||||
if update_item.shopping_list_id not in existing_items_map:
|
||||
query = PaginationQuery(
|
||||
per_page=-1, query_filter=f"shopping_list_id={update_item.shopping_list_id} AND checked=false"
|
||||
)
|
||||
items_data = self.list_items.page_all(query)
|
||||
existing_items_map[update_item.shopping_list_id] = items_data.items
|
||||
|
||||
merged = False
|
||||
for existing_item in existing_items_map[update_item.shopping_list_id]:
|
||||
if existing_item.id in delete_items or existing_item.id == update_item.id:
|
||||
continue
|
||||
|
||||
if not self.can_merge(update_item, existing_item):
|
||||
continue
|
||||
|
||||
updated_existing_item = self.merge_items(update_item, existing_item).cast(
|
||||
ShoppingListItemUpdateBulk, id=existing_item.id
|
||||
)
|
||||
filtered_update_items.append(updated_existing_item)
|
||||
delete_items.add(update_item.id)
|
||||
merged = True
|
||||
break
|
||||
|
||||
if merged:
|
||||
continue
|
||||
|
||||
# update or delete the item
|
||||
if update_item.quantity < 0:
|
||||
delete_items.add(update_item.id)
|
||||
continue
|
||||
|
||||
if update_item.checked:
|
||||
# checked items should not have recipe references
|
||||
update_item.recipe_references = []
|
||||
|
||||
filtered_update_items.append(update_item)
|
||||
|
||||
updated_items = cast(
|
||||
list[ShoppingListItemOut],
|
||||
self.list_items.update_many(filtered_update_items) if filtered_update_items else [], # type: ignore
|
||||
)
|
||||
|
||||
deleted_items = cast(
|
||||
list[ShoppingListItemOut],
|
||||
self.list_items.delete_many(delete_items) if delete_items else [], # type: ignore
|
||||
)
|
||||
|
||||
for list_id in set(item.shopping_list_id for item in updated_items + deleted_items):
|
||||
self.remove_unused_recipe_references(list_id)
|
||||
|
||||
return ShoppingListItemsCollectionOut(
|
||||
created_items=[], updated_items=updated_items, deleted_items=deleted_items
|
||||
)
|
||||
|
||||
def bulk_delete_items(self, delete_items: list[UUID4]) -> ShoppingListItemsCollectionOut:
|
||||
deleted_items = cast(
|
||||
list[ShoppingListItemOut],
|
||||
self.list_items.delete_many(set(delete_items)) if delete_items else [], # type: ignore
|
||||
)
|
||||
|
||||
for list_id in set(item.shopping_list_id for item in deleted_items):
|
||||
self.remove_unused_recipe_references(list_id)
|
||||
|
||||
return ShoppingListItemsCollectionOut(created_items=[], updated_items=[], deleted_items=deleted_items)
|
||||
|
||||
def get_shopping_list_items_from_recipe(
|
||||
self, list_id: UUID4, recipe_id: UUID4, scale: float = 1
|
||||
) -> list[ShoppingListItemCreate]:
|
||||
"""Generates a list of new list items based on a recipe"""
|
||||
|
||||
recipe = self.repos.recipes.get_one(recipe_id, "id")
|
||||
if not recipe:
|
||||
raise UnexpectedNone("Recipe not found")
|
||||
|
||||
to_create = []
|
||||
list_items: list[ShoppingListItemCreate] = []
|
||||
for ingredient in recipe.recipe_ingredient:
|
||||
food_id = None
|
||||
try:
|
||||
food_id = ingredient.food.id # type: ignore
|
||||
except AttributeError:
|
||||
pass
|
||||
if isinstance(ingredient.food, IngredientFood):
|
||||
is_food = True
|
||||
food_id = ingredient.food.id
|
||||
label_id = ingredient.food.label_id
|
||||
|
||||
label_id = None
|
||||
try:
|
||||
label_id = ingredient.food.label.id # type: ignore
|
||||
except AttributeError:
|
||||
pass
|
||||
else:
|
||||
is_food = False
|
||||
food_id = None
|
||||
label_id = None
|
||||
|
||||
unit_id = None
|
||||
try:
|
||||
unit_id = ingredient.unit.id # type: ignore
|
||||
except AttributeError:
|
||||
pass
|
||||
if isinstance(ingredient.unit, IngredientUnit):
|
||||
unit_id = ingredient.unit.id
|
||||
|
||||
to_create.append(
|
||||
ShoppingListItemCreate(
|
||||
shopping_list_id=list_id,
|
||||
is_food=not recipe.settings.disable_amount if recipe.settings else False,
|
||||
food_id=food_id,
|
||||
unit_id=unit_id,
|
||||
quantity=ingredient.quantity * recipe_increment if ingredient.quantity else 0,
|
||||
note=ingredient.note,
|
||||
label_id=label_id,
|
||||
recipe_id=recipe_id,
|
||||
recipe_references=[
|
||||
ShoppingListItemRecipeRef(
|
||||
recipe_id=recipe_id, recipe_quantity=ingredient.quantity, recipe_scale=recipe_increment
|
||||
)
|
||||
],
|
||||
)
|
||||
else:
|
||||
unit_id = None
|
||||
|
||||
new_item = ShoppingListItemCreate(
|
||||
shopping_list_id=list_id,
|
||||
is_food=is_food,
|
||||
note=ingredient.note,
|
||||
quantity=ingredient.quantity * scale if ingredient.quantity else 0,
|
||||
food_id=food_id,
|
||||
label_id=label_id,
|
||||
unit_id=unit_id,
|
||||
recipe_references=[
|
||||
ShoppingListItemRecipeRefCreate(
|
||||
recipe_id=recipe.id, recipe_quantity=ingredient.quantity, recipe_scale=scale
|
||||
)
|
||||
],
|
||||
)
|
||||
|
||||
new_shopping_list_items = [self.repos.group_shopping_list_item.create(item) for item in to_create]
|
||||
# some recipes have the same ingredient multiple times, so we check to see if we can combine them
|
||||
merged = False
|
||||
for existing_item in list_items:
|
||||
if not self.can_merge(existing_item, new_item):
|
||||
continue
|
||||
|
||||
updated_shopping_list = self.shopping_lists.get_one(list_id)
|
||||
if not updated_shopping_list:
|
||||
raise UnexpectedNone("Shopping List not found")
|
||||
# since this is the same recipe, we combine the quanities, rather than the scales
|
||||
# all items will have exactly one recipe reference
|
||||
if ingredient.quantity:
|
||||
existing_item.quantity += ingredient.quantity
|
||||
existing_item.recipe_references[0].recipe_quantity += ingredient.quantity # type: ignore
|
||||
|
||||
updated_shopping_list_items, deleted_shopping_list_items = self.consolidate_and_save(
|
||||
updated_shopping_list.list_items, # type: ignore
|
||||
)
|
||||
updated_shopping_list.list_items = updated_shopping_list_items
|
||||
# merge notes
|
||||
if existing_item.note != new_item.note:
|
||||
existing_item.note = " | ".join([note for note in [existing_item.note, new_item.note] if note])
|
||||
|
||||
not_found = True
|
||||
for refs in updated_shopping_list.recipe_references:
|
||||
if refs.recipe_id != recipe_id:
|
||||
merged = True
|
||||
break
|
||||
|
||||
if not merged:
|
||||
list_items.append(new_item)
|
||||
|
||||
return list_items
|
||||
|
||||
def add_recipe_ingredients_to_list(
|
||||
self, list_id: UUID4, recipe_id: UUID4, recipe_increment: float = 1
|
||||
) -> tuple[ShoppingListOut, ShoppingListItemsCollectionOut]:
|
||||
"""
|
||||
Adds a recipe's ingredients to a list
|
||||
|
||||
Returns a tuple of:
|
||||
- Updated Shopping List
|
||||
- Impacted Shopping List Items
|
||||
"""
|
||||
|
||||
items_to_create = self.get_shopping_list_items_from_recipe(list_id, recipe_id, recipe_increment)
|
||||
item_changes = self.bulk_create_items(items_to_create)
|
||||
|
||||
updated_list = cast(ShoppingListOut, self.shopping_lists.get_one(list_id))
|
||||
|
||||
ref_merged = False
|
||||
for ref in updated_list.recipe_references:
|
||||
if ref.recipe_id != recipe_id:
|
||||
continue
|
||||
|
||||
refs.recipe_quantity += recipe_increment
|
||||
not_found = False
|
||||
ref.recipe_quantity += recipe_increment
|
||||
ref_merged = True
|
||||
break
|
||||
|
||||
if not_found:
|
||||
updated_shopping_list.recipe_references.append(
|
||||
ShoppingListItemRecipeRef(recipe_id=recipe_id, recipe_quantity=recipe_increment) # type: ignore
|
||||
if not ref_merged:
|
||||
updated_list.recipe_references.append(
|
||||
ShoppingListItemRecipeRefCreate(recipe_id=recipe_id, recipe_quantity=recipe_increment) # type: ignore
|
||||
)
|
||||
|
||||
updated_shopping_list = self.shopping_lists.update(updated_shopping_list.id, updated_shopping_list)
|
||||
|
||||
"""
|
||||
There can be overlap between the list item collections, so we de-duplicate the lists.
|
||||
|
||||
First new items are created, then existing items are updated, and finally some items are deleted,
|
||||
so we can de-duplicate using this logic
|
||||
"""
|
||||
new_items_map = {list_item.id: list_item for list_item in new_shopping_list_items}
|
||||
updated_items_map = {list_item.id: list_item for list_item in updated_shopping_list_items}
|
||||
deleted_items_map = {list_item.id: list_item for list_item in deleted_shopping_list_items}
|
||||
|
||||
# if the item was created and then updated, replace the create with the update and remove the update
|
||||
for id in list(updated_items_map.keys()):
|
||||
if id in new_items_map:
|
||||
new_items_map[id] = updated_items_map[id]
|
||||
del updated_items_map[id]
|
||||
|
||||
# if the item was updated and then deleted, remove the update
|
||||
updated_shopping_list_items = [
|
||||
list_item for id, list_item in updated_items_map.items() if id not in deleted_items_map
|
||||
]
|
||||
|
||||
# if the item was created and then deleted, remove it from both lists
|
||||
new_shopping_list_items = [list_item for id, list_item in new_items_map.items() if id not in deleted_items_map]
|
||||
deleted_shopping_list_items = [
|
||||
list_item for id, list_item in deleted_items_map.items() if id not in new_items_map
|
||||
]
|
||||
|
||||
return updated_shopping_list, new_shopping_list_items, updated_shopping_list_items, deleted_shopping_list_items
|
||||
updated_list = self.shopping_lists.update(updated_list.id, updated_list)
|
||||
return updated_list, item_changes
|
||||
|
||||
def remove_recipe_ingredients_from_list(
|
||||
self, list_id: UUID4, recipe_id: UUID4, recipe_decrement: float = 1
|
||||
) -> tuple[ShoppingListOut, list[ShoppingListItemOut], list[ShoppingListItemOut]]:
|
||||
) -> tuple[ShoppingListOut, ShoppingListItemsCollectionOut]:
|
||||
"""
|
||||
returns:
|
||||
- updated_shopping_list
|
||||
- updated_shopping_list_items
|
||||
- deleted_shopping_list_items
|
||||
Removes a recipe's ingredients from a list
|
||||
|
||||
Returns a tuple of:
|
||||
- Updated Shopping List
|
||||
- Impacted Shopping List Items
|
||||
"""
|
||||
|
||||
shopping_list = self.shopping_lists.get_one(list_id)
|
||||
if shopping_list is None:
|
||||
raise UnexpectedNone("Shopping list not found, cannot remove recipe ingredients")
|
||||
|
||||
updated_shopping_list_items = []
|
||||
deleted_shopping_list_items = []
|
||||
update_items: list[ShoppingListItemUpdateBulk] = []
|
||||
delete_items: list[UUID4] = []
|
||||
for item in shopping_list.list_items:
|
||||
found = False
|
||||
|
||||
@@ -270,10 +393,6 @@ class ShoppingListService:
|
||||
if ref.recipe_scale is None:
|
||||
ref.recipe_scale = 1
|
||||
|
||||
# recipe quantity should never be None, but we check just in case
|
||||
if ref.recipe_quantity is None:
|
||||
ref.recipe_quantity = 0
|
||||
|
||||
# Set Quantity
|
||||
if ref.recipe_scale > recipe_decrement:
|
||||
# remove only part of the reference
|
||||
@@ -286,25 +405,33 @@ class ShoppingListService:
|
||||
# Set Reference Scale
|
||||
ref.recipe_scale -= recipe_decrement
|
||||
if ref.recipe_scale <= 0:
|
||||
# delete the ref from the database and remove it from our list
|
||||
self.list_item_refs.delete(ref.id)
|
||||
item.recipe_references.remove(ref)
|
||||
|
||||
found = True
|
||||
break
|
||||
|
||||
# If the item was found we need to check its new quantity
|
||||
if found:
|
||||
if item.quantity <= 0:
|
||||
self.list_items.delete(item.id)
|
||||
deleted_shopping_list_items.append(item)
|
||||
# only remove a 0 quantity item if we removed its last recipe reference
|
||||
if item.quantity < 0 or (item.quantity == 0 and not item.recipe_references):
|
||||
delete_items.append(item.id)
|
||||
|
||||
else:
|
||||
self.list_items.update(item.id, item)
|
||||
updated_shopping_list_items.append(item)
|
||||
update_items.append(item.cast(ShoppingListItemUpdateBulk))
|
||||
|
||||
response_update = self.bulk_update_items(update_items)
|
||||
|
||||
deleted_item_ids = [item.id for item in response_update.deleted_items]
|
||||
response_delete = self.bulk_delete_items([id for id in delete_items if id not in deleted_item_ids])
|
||||
|
||||
items = ShoppingListItemsCollectionOut(
|
||||
created_items=response_update.created_items + response_delete.created_items,
|
||||
updated_items=response_update.updated_items + response_delete.updated_items,
|
||||
deleted_items=response_update.deleted_items + response_delete.deleted_items,
|
||||
)
|
||||
|
||||
# Decrement the list recipe reference count
|
||||
for recipe_ref in shopping_list.recipe_references:
|
||||
updated_list = self.shopping_lists.get_one(shopping_list.id)
|
||||
for recipe_ref in updated_list.recipe_references: # type: ignore
|
||||
if recipe_ref.recipe_id != recipe_id or recipe_ref.recipe_quantity is None:
|
||||
continue
|
||||
|
||||
@@ -318,8 +445,4 @@ class ShoppingListService:
|
||||
|
||||
break
|
||||
|
||||
return (
|
||||
self.shopping_lists.get_one(shopping_list.id),
|
||||
updated_shopping_list_items,
|
||||
deleted_shopping_list_items,
|
||||
) # type: ignore
|
||||
return self.shopping_lists.get_one(shopping_list.id), items # type: ignore
|
||||
|
||||
@@ -4,6 +4,7 @@ from fractions import Fraction
|
||||
from mealie.core.root_logger import get_logger
|
||||
from mealie.schema.recipe import RecipeIngredient
|
||||
from mealie.schema.recipe.recipe_ingredient import (
|
||||
MAX_INGREDIENT_DENOMINATOR,
|
||||
CreateIngredientFood,
|
||||
CreateIngredientUnit,
|
||||
IngredientConfidence,
|
||||
@@ -74,7 +75,9 @@ class NLPParser(ABCIngredientParser):
|
||||
unit=CreateIngredientUnit(name=crf_model.unit),
|
||||
food=CreateIngredientFood(name=crf_model.name),
|
||||
disable_amount=False,
|
||||
quantity=float(sum(Fraction(s).limit_denominator(32) for s in crf_model.qty.split())),
|
||||
quantity=float(
|
||||
sum(Fraction(s).limit_denominator(MAX_INGREDIENT_DENOMINATOR) for s in crf_model.qty.split())
|
||||
),
|
||||
)
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to parse ingredient: {crf_model}: {e}")
|
||||
|
||||
Reference in New Issue
Block a user