munch-ease-backend/shopping/__init__.py

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,
    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,
    )