from __future__ import annotations from typing import Dict import aiosqlite from fastapi import APIRouter, Depends import shopping from api.deps import get_db, get_household_from_slug from api.shopping import CurrentShoppingList, _to_ingredient_item, _to_meal_item, _to_shopping_list_out router = APIRouter(prefix="/households/{householdSlug}/shopping", tags=["shopping-v2"]) @router.get( "/current", response_model=CurrentShoppingList, operation_id="getCurrentShoppingListV2", summary="Get the current aggregated shopping list (scoped)", ) async def get_current_shopping_list_scoped( household=Depends(get_household_from_slug), conn: aiosqlite.Connection = Depends(get_db), ) -> CurrentShoppingList: hid = household["id"] ( outstanding_requests, purchased_requests, meal_requests, meals_lookup, recipes_lookup, ingredients_lookup, ) = await shopping.get_outstanding_requests_scoped(conn, hid) # Load full lists for additional lookups (by household) other_shopping_list_ids = {item.list_id for item in purchased_requests} other_lists_domain: Dict[int, shopping.ShoppingList] = {} for list_id in other_shopping_list_ids: if list_id is not None: sl = await shopping.load_shopping_list_scoped(conn, list_id, hid) if sl is not None: other_lists_domain[list_id] = sl # Add any additional items from shopping lists to the existing lookups additional_items = [item for sl in other_lists_domain.values() for item in sl.items] if additional_items: await shopping.to_lookups(conn, additional_items, meals_lookup, recipes_lookup, ingredients_lookup) shopping_list_lookup = {k: _to_shopping_list_out(v) for k, v in other_lists_domain.items()} return CurrentShoppingList( outstanding_items=[_to_ingredient_item(i) for i in outstanding_requests], requested_meals=[_to_meal_item(i) for i in meal_requests], purchased_items=[_to_ingredient_item(i) for i in purchased_requests], meals_lookup=meals_lookup, shopping_list_lookup=shopping_list_lookup, ingredients_lookup=ingredients_lookup, recipes_lookup=recipes_lookup, )