from typing import Any, Dict, Iterable, Iterator, List, Tuple import ingredients import meals import recipes from shopping.models import ShoppingList as ShoppingList, ShoppingListItem as ShoppingListItem from shopping.repository import ( find_items_by_list_id as _find_items_by_list_id, find_items_by_list_id_scoped as _find_items_by_list_id_scoped, get_purchased_ingredients as _get_purchased_ingredients, get_purchased_ingredients_scoped as _get_purchased_ingredients_scoped, is_requested as is_requested, request_meal_scoped as request_meal_scoped, request_ingredient_scoped as request_ingredient_scoped, remove_meal_request_scoped as remove_meal_request_scoped, load_shopping_list as load_shopping_list, load_shopping_list_scoped as load_shopping_list_scoped, purchase as purchase, remove_request as remove_request, request as request, update_purchased_meals as update_purchased_meals, validate_request as validate_request, ) async def to_lookups( conn, items: List[ShoppingListItem], meals_lookup: Dict[int, Any] | None = None, recipes_lookup: Dict[int, Any] | None = None, ingredients_lookup: Dict[int, Any] | None = None, ) -> Tuple[Dict[int, Any], Dict[int, Any], Dict[int, Any]]: meals_lookup = meals_lookup or {} recipes_lookup = recipes_lookup or {} ingredients_lookup = ingredients_lookup or {} await _ensure_lookups_populated(conn, items, meals_lookup, recipes_lookup, ingredients_lookup) return meals_lookup, recipes_lookup, ingredients_lookup async def _ensure_lookups_populated( conn, items: List[ShoppingListItem], meals_lookup, recipes_lookup, ingredients_lookup ): for item in items: # If the any item is not in the lookup, we need to add it if item.meal_id and item.meal_id not in meals_lookup: meals_lookup[item.meal_id] = await meals.find_meal_by_id(conn, item.meal_id) if item.recipe_id and item.recipe_id not in recipes_lookup: recipes_lookup[item.recipe_id] = await recipes.find_recipe_by_id(conn, item.recipe_id) if item.ingredient_id and item.ingredient_id not in ingredients_lookup: ingredients_lookup[item.ingredient_id] = await ingredients.find_ingredient_by_id( conn, item.ingredient_id ) async def get_persons_requests(conn, person_id: int) -> List[ingredients.Ingredient]: ids = [ item.ingredient_id async for item in _find_items_by_list_id(conn, None) if item.person_id == person_id and item.ingredient_id is not None and item.meal_id is None ] return [ ing for ing in [ await ingredients.find_ingredient_by_id(conn, ingredient_id) for ingredient_id in ids ] if ing is not None ] def flatten_items( items: Iterable[ShoppingListItem], meals_lookup: Dict[int, Any] ) -> Iterator[ShoppingListItem]: for item in items: if item.meal_id and item.meal_id in meals_lookup: meal = meals_lookup[item.meal_id] for mealRecipe in meal.recipes: for ingredient in mealRecipe.recipe.ingredients: yield ShoppingListItem( ingredient_id=ingredient.id, meal_id=item.meal_id, recipe_id=mealRecipe.recipe.id, person_id=item.person_id, created_date=item.created_date, ) for ingredient in meal.extra_ingredients: yield ShoppingListItem( ingredient_id=ingredient.id, meal_id=item.meal_id, person_id=item.person_id, created_date=item.created_date, ) else: yield item async def get_outstanding_requests( conn, ) -> Tuple[ List[ShoppingListItem], List[ShoppingListItem], List[ShoppingListItem], Dict[int, Any], Dict[int, Any], Dict[int, Any], ]: current_requests = [r async for r in _find_items_by_list_id(conn, None)] meal_requests = [r for r in current_requests if r.meal_id is not None and r.meal_id > 0] # Get lookups for meals to enable flattening meals_lookup, recipes_lookup, ingredients_lookup = await to_lookups(conn, current_requests) meal_ids = [r.meal_id for r in meal_requests if r.meal_id] purchased_ingredients = { (r.ingredient_id, r.meal_id, r.recipe_id): r async for r in _get_purchased_ingredients(conn, meal_ids) } outstanding_items = [] purchased_items = [] flattened = list(flatten_items(current_requests, meals_lookup)) # Now ensure that all ingredients from the flattened items are in the lookup await _ensure_lookups_populated( conn, flattened, meals_lookup, recipes_lookup, ingredients_lookup ) for r in flattened: # Meal ingredients may have already been purchased if r.meal_id is not None and r.meal_id > 0: purchased_item = purchased_ingredients.get((r.ingredient_id, r.meal_id, r.recipe_id)) if purchased_item: purchased_items.append(purchased_item) continue outstanding_items.append(r) return ( outstanding_items, purchased_items, meal_requests, meals_lookup, recipes_lookup, ingredients_lookup, ) async def get_outstanding_requests_scoped( conn, household_id: int, ) -> Tuple[ List[ShoppingListItem], List[ShoppingListItem], List[ShoppingListItem], Dict[int, Any], Dict[int, Any], Dict[int, Any], ]: current_requests = [r async for r in _find_items_by_list_id_scoped(conn, None, household_id)] meal_requests = [r for r in current_requests if r.meal_id is not None and r.meal_id > 0] # Get lookups for meals to enable flattening meals_lookup, recipes_lookup, ingredients_lookup = await to_lookups(conn, current_requests) meal_ids = [r.meal_id for r in meal_requests if r.meal_id] purchased_ingredients = { (r.ingredient_id, r.meal_id, r.recipe_id): r async for r in _get_purchased_ingredients_scoped(conn, meal_ids, household_id) } outstanding_items = [] purchased_items = [] flattened = list(flatten_items(current_requests, meals_lookup)) # Now ensure that all ingredients from the flattened items are in the lookup await _ensure_lookups_populated( conn, flattened, meals_lookup, recipes_lookup, ingredients_lookup ) for r in flattened: # Meal ingredients may have already been purchased (by list in the same household) if r.meal_id is not None and r.meal_id > 0: purchased_item = purchased_ingredients.get((r.ingredient_id, r.meal_id, r.recipe_id)) if purchased_item: purchased_items.append(purchased_item) continue outstanding_items.append(r) return ( outstanding_items, purchased_items, meal_requests, meals_lookup, recipes_lookup, ingredients_lookup, )