143 lines
4.9 KiB
Python
143 lines
4.9 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Dict, List
|
|
|
|
import aiosqlite
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
|
|
import shopping
|
|
from api.deps import error_response, get_db, get_household_from_slug
|
|
from api.shopping import (
|
|
CurrentShoppingList,
|
|
PurchasedShoppingList,
|
|
_to_ingredient_item,
|
|
_to_meal_item,
|
|
_to_shopping_list_out,
|
|
IngredientPurchaseItemIn,
|
|
PurchaseListIn,
|
|
)
|
|
|
|
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,
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/{list_id}",
|
|
response_model=PurchasedShoppingList,
|
|
operation_id="getShoppingListV2",
|
|
summary="Get a purchased shopping list by id (scoped)",
|
|
)
|
|
async def get_shopping_list_scoped(
|
|
list_id: int,
|
|
request: Request,
|
|
household=Depends(get_household_from_slug),
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
) -> PurchasedShoppingList | Response:
|
|
hid = household["id"]
|
|
shopping_list = await shopping.load_shopping_list_scoped(conn, list_id, hid)
|
|
if not shopping_list:
|
|
return error_response(request, 404, "Shopping list not found")
|
|
|
|
meals_lookup, recipes_lookup, ingredients_lookup = await shopping.to_lookups(
|
|
conn, shopping_list.items
|
|
)
|
|
return PurchasedShoppingList(
|
|
list=_to_shopping_list_out(shopping_list),
|
|
meals_lookup=meals_lookup,
|
|
recipes_lookup=recipes_lookup,
|
|
ingredients_lookup=ingredients_lookup,
|
|
)
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=PurchasedShoppingList,
|
|
operation_id="purchaseIngredientsV2",
|
|
summary="Purchase ingredients for a shopping list (scoped)",
|
|
)
|
|
async def purchase_ingredients_scoped(
|
|
shopping_list: PurchaseListIn,
|
|
request: Request,
|
|
household=Depends(get_household_from_slug),
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
) -> PurchasedShoppingList | Response:
|
|
hid = household["id"]
|
|
|
|
# Map outward input DTO to domain model
|
|
domain_items: List[shopping.ShoppingListItem] = []
|
|
for it in shopping_list.items:
|
|
created = it.created_date or __import__("datetime").datetime.now().astimezone()
|
|
domain_items.append(
|
|
shopping.ShoppingListItem(
|
|
ingredient_id=it.ingredient_id,
|
|
person_id=it.person_id,
|
|
meal_id=it.meal_id,
|
|
recipe_id=it.recipe_id,
|
|
created_date=created,
|
|
)
|
|
)
|
|
|
|
domain_list = shopping.ShoppingList(items=domain_items, store_name=shopping_list.store_name)
|
|
|
|
try:
|
|
# Use household-scoped purchase which ensures requests belong to the same household
|
|
await shopping.purchase_scoped(conn, domain_list, hid)
|
|
except ValueError as e:
|
|
return error_response(request, 400, str(e))
|
|
|
|
# Lookups for outward response
|
|
meals_lookup, recipes_lookup, ingredients_lookup = await shopping.to_lookups(
|
|
conn, domain_list.items
|
|
)
|
|
return PurchasedShoppingList(
|
|
list=_to_shopping_list_out(domain_list),
|
|
meals_lookup=meals_lookup,
|
|
recipes_lookup=recipes_lookup,
|
|
ingredients_lookup=ingredients_lookup,
|
|
)
|