361 lines
11 KiB
Python
361 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from datetime import datetime
|
|
from typing import Dict, List, Literal
|
|
|
|
import aiosqlite
|
|
from fastapi import APIRouter, Depends, Request, Response
|
|
|
|
import ingredients
|
|
import meals
|
|
import persons
|
|
import recipes
|
|
import shopping
|
|
from shopping.models import StoreEnum
|
|
from api.deps import cookie_person, error_response, get_db
|
|
from common import ApiModel, Field, ProblemDetails
|
|
|
|
|
|
# Outward-facing models to reduce unnecessary nulls in API responses
|
|
class ListIngredientItem(ApiModel):
|
|
kind: Literal["ingredient"] = "ingredient"
|
|
id: int = -1
|
|
ingredient_id: int
|
|
person_id: int
|
|
created_date: datetime
|
|
# These may be present when the ingredient is part of a requested meal
|
|
list_id: int | None = None
|
|
meal_id: int | None = None
|
|
recipe_id: int | None = None
|
|
|
|
|
|
class RequestedMealItem(ApiModel):
|
|
kind: Literal["requestedMeal"] = "requestedMeal"
|
|
id: int = -1
|
|
person_id: int
|
|
meal_id: int
|
|
created_date: datetime
|
|
|
|
|
|
# Input DTOs (separate from internal DB/domain models)
|
|
class IngredientPurchaseItemIn(ApiModel):
|
|
ingredient_id: int
|
|
person_id: int
|
|
created_date: datetime | None = None
|
|
meal_id: int | None = None
|
|
recipe_id: int | None = None
|
|
|
|
|
|
class PurchaseListIn(ApiModel):
|
|
store_name: StoreEnum
|
|
items: List[IngredientPurchaseItemIn]
|
|
|
|
|
|
# Output DTOs for purchased lists
|
|
class ShoppingListOut(ApiModel):
|
|
id: int
|
|
created_date: datetime
|
|
store_name: StoreEnum
|
|
purchased_by_id: int
|
|
purchased_by: persons.Person | None = None
|
|
items: List[ListIngredientItem] = Field(default_factory=list)
|
|
|
|
|
|
router = APIRouter(prefix="/shopping", tags=["shopping"])
|
|
|
|
|
|
class CurrentShoppingList(ApiModel):
|
|
outstanding_items: List[ListIngredientItem]
|
|
requested_meals: List[RequestedMealItem]
|
|
purchased_items: List[ListIngredientItem] = Field(default_factory=list)
|
|
|
|
ingredients_lookup: Dict[int, ingredients.Ingredient] = Field(default_factory=dict)
|
|
meals_lookup: Dict[int, meals.Meal] = Field(default_factory=dict)
|
|
shopping_list_lookup: Dict[int, ShoppingListOut] = Field(default_factory=dict)
|
|
recipes_lookup: Dict[int, recipes.Recipe] = Field(default_factory=dict)
|
|
|
|
|
|
# Mapping helpers from domain -> outward API
|
|
def _to_ingredient_item(item: shopping.ShoppingListItem) -> ListIngredientItem:
|
|
return ListIngredientItem(
|
|
id=item.id,
|
|
ingredient_id=item.ingredient_id if item.ingredient_id is not None else -1,
|
|
person_id=item.person_id,
|
|
created_date=item.created_date,
|
|
list_id=item.list_id,
|
|
meal_id=item.meal_id,
|
|
recipe_id=item.recipe_id,
|
|
)
|
|
|
|
|
|
def _to_meal_item(item: shopping.ShoppingListItem) -> RequestedMealItem:
|
|
return RequestedMealItem(
|
|
id=item.id,
|
|
person_id=item.person_id,
|
|
meal_id=item.meal_id if item.meal_id is not None else -1,
|
|
created_date=item.created_date,
|
|
)
|
|
|
|
|
|
def _to_shopping_list_out(sl: shopping.ShoppingList) -> ShoppingListOut:
|
|
return ShoppingListOut(
|
|
id=sl.id,
|
|
created_date=sl.created_date,
|
|
store_name=sl.store_name,
|
|
purchased_by_id=sl.purchased_by_id,
|
|
purchased_by=sl.purchased_by,
|
|
items=[_to_ingredient_item(i) for i in sl.items],
|
|
)
|
|
|
|
|
|
@router.get(
|
|
"/current",
|
|
response_model=CurrentShoppingList,
|
|
operation_id="getCurrentShoppingList",
|
|
summary="Get the current aggregated shopping list",
|
|
)
|
|
async def get_current_shopping_list(
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
) -> CurrentShoppingList:
|
|
(
|
|
outstanding_requests,
|
|
purchased_requests,
|
|
meal_requests,
|
|
meals_lookup,
|
|
recipes_lookup,
|
|
ingredients_lookup,
|
|
) = await shopping.get_outstanding_requests(conn)
|
|
other_shopping_list_ids = {item.list_id for item in purchased_requests}
|
|
|
|
# Load full lists for additional lookups
|
|
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(conn, list_id)
|
|
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
|
|
)
|
|
|
|
# Convert domain shopping lists to outward form for response
|
|
shopping_list_lookup: Dict[int, ShoppingListOut] = {
|
|
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,
|
|
)
|
|
|
|
|
|
class PurchasedShoppingList(ApiModel):
|
|
list: ShoppingListOut
|
|
meals_lookup: Dict[int, meals.Meal] = Field(default_factory=dict)
|
|
ingredients_lookup: Dict[int, ingredients.Ingredient] = Field(default_factory=dict)
|
|
recipes_lookup: Dict[int, recipes.Recipe] = Field(default_factory=dict)
|
|
|
|
|
|
@router.get(
|
|
"/{list_id}",
|
|
response_model=PurchasedShoppingList,
|
|
operation_id="getShoppingList",
|
|
summary="Get a purchased shopping list by id",
|
|
responses={
|
|
404: {
|
|
"model": ProblemDetails,
|
|
"description": "Shopping list not found",
|
|
"content": {"application/problem+json": {}},
|
|
}
|
|
},
|
|
)
|
|
async def get_shopping_list(
|
|
list_id: int, request: Request, conn: aiosqlite.Connection = Depends(get_db)
|
|
) -> PurchasedShoppingList | Response:
|
|
shopping_list = await shopping.load_shopping_list(conn, list_id)
|
|
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="purchaseIngredients",
|
|
summary="Purchase ingredients for a shopping list",
|
|
responses={
|
|
400: {
|
|
"model": ProblemDetails,
|
|
"description": "Validation error",
|
|
"content": {"application/problem+json": {}},
|
|
},
|
|
401: {
|
|
"model": ProblemDetails,
|
|
"description": "Unauthorized (invalid or unknown user)",
|
|
"content": {"application/problem+json": {}},
|
|
},
|
|
},
|
|
)
|
|
async def purchase_ingredients(
|
|
shopping_list: PurchaseListIn,
|
|
request: Request,
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
person: persons.Person = Depends(cookie_person),
|
|
) -> PurchasedShoppingList | Response:
|
|
# Ensure the caller is authenticated and maps to a known user
|
|
if not person:
|
|
return error_response(request, 401, "Unauthorized")
|
|
|
|
# Map outward input DTO to domain model
|
|
domain_items: List[shopping.ShoppingListItem] = []
|
|
for it in shopping_list.items:
|
|
created = it.created_date or 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(
|
|
purchased_by=person, items=domain_items, store_name=shopping_list.store_name
|
|
)
|
|
try:
|
|
await shopping.purchase(conn, domain_list)
|
|
except ValueError as e:
|
|
# Map domain validation errors to a proper Problem Details response
|
|
return error_response(request, 400, str(e))
|
|
result = PurchasedShoppingList(list=_to_shopping_list_out(domain_list))
|
|
await shopping.to_lookups(
|
|
conn,
|
|
domain_list.items,
|
|
result.meals_lookup,
|
|
result.recipes_lookup,
|
|
result.ingredients_lookup,
|
|
)
|
|
return result
|
|
|
|
|
|
@router.get(
|
|
"/current/me/ingredients",
|
|
operation_id="getMyShoppingList",
|
|
summary="Get my outstanding ingredient requests",
|
|
)
|
|
async def get_my_shopping_list(
|
|
conn: aiosqlite.Connection = Depends(get_db), person: persons.Person = Depends(cookie_person)
|
|
) -> List[ingredients.Ingredient]:
|
|
return await shopping.get_persons_requests(conn, person.id)
|
|
|
|
|
|
@router.post(
|
|
"/current/me/ingredients",
|
|
operation_id="syncMyShoppingList",
|
|
summary="Sync my outstanding ingredient requests",
|
|
)
|
|
async def sync_my_shopping_list(
|
|
requests: List[ingredients.Ingredient],
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
person: persons.Person = Depends(cookie_person),
|
|
) -> List[ingredients.Ingredient]:
|
|
def isMatching(a: ingredients.Ingredient, b: ingredients.Ingredient) -> bool:
|
|
return a.id == b.id or a.line == b.line
|
|
|
|
my_shopping_list = await shopping.get_persons_requests(conn, person.id)
|
|
to_remove = [r for r in my_shopping_list if not any(isMatching(r, req) for req in requests)]
|
|
to_add = [req for req in requests if not any(isMatching(req, r) for r in my_shopping_list)]
|
|
|
|
for r in to_remove:
|
|
await shopping.remove_request(conn, person, ingredient=r)
|
|
|
|
for r in to_add:
|
|
if r.id < 0:
|
|
await ingredients.insert_ingredient(conn, r)
|
|
await shopping.request(conn, person, ingredient=r)
|
|
|
|
return await get_my_shopping_list(conn, person)
|
|
|
|
|
|
class MealIdWrapper(ApiModel):
|
|
meal_id: int
|
|
|
|
|
|
@router.post(
|
|
"/current/meals/me",
|
|
response_model=RequestedMealItem,
|
|
operation_id="requestMeal",
|
|
summary="Request a meal for shopping",
|
|
responses={
|
|
404: {
|
|
"model": ProblemDetails,
|
|
"description": "Meal not found",
|
|
"content": {"application/problem+json": {}},
|
|
}
|
|
},
|
|
)
|
|
async def request_meal(
|
|
r: MealIdWrapper,
|
|
request: Request,
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
person: persons.Person = Depends(cookie_person),
|
|
) -> RequestedMealItem | Response:
|
|
meal = await meals.find_meal_by_id(conn, r.meal_id)
|
|
if not meal:
|
|
return error_response(request, 404, "Meal not found")
|
|
|
|
response = await shopping.request(conn, person, meal=meal)
|
|
return _to_meal_item(response)
|
|
|
|
|
|
class Ok(ApiModel):
|
|
ok: bool = True
|
|
|
|
|
|
@router.delete(
|
|
"/current/meals/{meal_id}",
|
|
response_model=Ok,
|
|
operation_id="unrequestMeal",
|
|
summary="Remove a meal request",
|
|
responses={
|
|
404: {
|
|
"model": ProblemDetails,
|
|
"description": "Meal not found",
|
|
"content": {"application/problem+json": {}},
|
|
}
|
|
},
|
|
)
|
|
async def unrequest_meal(
|
|
meal_id: int,
|
|
request: Request,
|
|
conn: aiosqlite.Connection = Depends(get_db),
|
|
person: persons.Person = Depends(cookie_person),
|
|
) -> Ok | Response:
|
|
meal = await meals.find_meal_by_id(conn, meal_id)
|
|
if not meal:
|
|
return error_response(request, 404, "Meal not found")
|
|
|
|
await shopping.remove_request(conn, person, meal=meal)
|
|
return Ok()
|
|
|
|
|
|
# Removed duplicate placeholder endpoints left over from earlier scaffolding
|