258 lines
No EOL
10 KiB
Python
258 lines
No EOL
10 KiB
Python
from common import BaseLinkedModel
|
|
from recipes import Recipe
|
|
from meals import Meal, find_meal_by_id
|
|
from ingredients import Ingredient, insert_ingredient
|
|
from persons import Person
|
|
from products import Product
|
|
|
|
from typing import AsyncIterator, List, ClassVar, Optional
|
|
|
|
from datetime import datetime
|
|
|
|
class ShoppingListItem(BaseLinkedModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'ingredient_id', 'list_id', 'person_id', 'meal_id', 'created_date']
|
|
id: int = -1
|
|
list_id: Optional[int] = None
|
|
|
|
person_id: int = -1
|
|
person: Optional[Person] = None
|
|
|
|
ingredient_id: Optional[int] = None
|
|
ingredient: Optional[Ingredient] = None
|
|
|
|
recipe_id: Optional[int] = None
|
|
recipe: Optional[Recipe] = None
|
|
|
|
meal_id: Optional[int] = None
|
|
meal: Optional[Meal] = None
|
|
|
|
created_date: datetime = datetime.now().astimezone()
|
|
|
|
|
|
from enum import Enum
|
|
|
|
class StoreEnum(str, Enum):
|
|
woolworths = 'woolworths'
|
|
coles = 'coles'
|
|
home = ''
|
|
|
|
class ShoppingList(BaseLinkedModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'created_date', 'store_name']
|
|
id: int = -1
|
|
created_date: datetime = datetime.now().astimezone()
|
|
store_name: StoreEnum = ''
|
|
purchased_by_id: int = -1
|
|
purchased_by: Optional[Person] = None
|
|
items: List[ShoppingListItem] = []
|
|
|
|
async def create(conn):
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS ShoppingList (
|
|
id INTEGER PRIMARY KEY,
|
|
created_date DATETIME NOT NULL,
|
|
store_name TEXT NOT NULL,
|
|
purchased_by_id INTEGER,
|
|
FOREIGN KEY(purchased_by_id) REFERENCES Person(id)
|
|
);''')
|
|
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS ShoppingListItem (
|
|
id INTEGER PRIMARY KEY,
|
|
ingredient_id INTEGER,
|
|
list_id INTEGER,
|
|
person_id INTEGER,
|
|
meal_id INTEGER,
|
|
recipe_id INTEGER,
|
|
created_date DATETIME NOT NULL,
|
|
FOREIGN KEY(ingredient_id) REFERENCES Ingredient(id),
|
|
FOREIGN KEY(list_id) REFERENCES ShoppingList(id),
|
|
FOREIGN KEY(person_id) REFERENCES Person(id),
|
|
FOREIGN KEY(meal_id) REFERENCES Meal(id),
|
|
FOREIGN KEY(recipe_id) REFERENCES Recipe(id)
|
|
);''')
|
|
|
|
|
|
def validate_request(request: ShoppingListItem) -> None:
|
|
if request.person_id < 0:
|
|
raise ValueError('Requests must have a person')
|
|
|
|
# A request must have either an ingredient or a meal, but not both
|
|
if not request.ingredient and not request.meal:
|
|
raise ValueError('Request must have either an ingredient or a meal')
|
|
|
|
async def purchase(conn, shopping_list: ShoppingList) -> None:
|
|
if shopping_list.purchased_by_id is None:
|
|
raise ValueError('Shopping list must have a person id')
|
|
|
|
if shopping_list.items is None or len(shopping_list.items) == 0:
|
|
raise ValueError('Shopping list must have items')
|
|
|
|
if shopping_list.purchased_by_id < 0:
|
|
raise ValueError('Shopping list must have a valid person id')
|
|
|
|
shopping_list.created_date = datetime.now().astimezone()
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingList (created_date, store_name, purchased_by_id)
|
|
VALUES (?, ?, ?)
|
|
''', (shopping_list.created_date.isoformat(), shopping_list.store_name, shopping_list.purchased_by_id)) as cursor:
|
|
shopping_list.id = cursor.lastrowid
|
|
|
|
for item in shopping_list.items:
|
|
item.list_id = shopping_list.id
|
|
validate_request(item)
|
|
|
|
if item.ingredient and item.ingredient.id < 0:
|
|
await insert_ingredient(conn, item.ingredient)
|
|
|
|
if item.ingredient_id is None or item.ingredient_id < 0:
|
|
raise ValueError('Ingredient request must have a valid ingredient id')
|
|
|
|
isMeal = item.meal_id is not None and item.meal_id >= 0
|
|
isPersonRequest = (not isMeal) and item.person_id is not None and item.person_id >= 0
|
|
|
|
if not isMeal and not isPersonRequest:
|
|
raise ValueError('Ingredient request must have either a meal or a person id')
|
|
|
|
if isPersonRequest:
|
|
# Update existing request from its null id, or throw
|
|
async with conn.execute('''
|
|
UPDATE ShoppingListItem
|
|
SET list_id = ?
|
|
WHERE ingredient_id = ?
|
|
AND list_id IS NULL
|
|
AND person_id = ?
|
|
AND meal_id IS NULL
|
|
AND recipe_id IS NULL
|
|
''', (shopping_list.id, item.ingredient_id, item.person_id)) as cursor:
|
|
if cursor.rowcount == 0:
|
|
raise ValueError('Ingredient request must have a valid person id and ingredient id')
|
|
|
|
elif isMeal:
|
|
# Insert new request for meal
|
|
if item.meal_id is None or item.meal_id < 0:
|
|
raise ValueError('Meal request must have a valid meal id')
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListItem (ingredient_id, list_id, person_id, meal_id, recipe_id, created_date)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
''', (item.ingredient_id, shopping_list.id, item.person_id, item.meal_id, item.recipe_id, item.created_date.isoformat())) as cursor:
|
|
item.id = cursor.lastrowid
|
|
|
|
# TODO: Calculate which meals have been fulfilled and update meal status
|
|
|
|
async def request(conn, person: Person, ingredient: Optional[Ingredient] = None, meal: Optional[Meal] = None) -> ShoppingListItem:
|
|
if ingredient is not None and meal is not None:
|
|
raise ValueError('Cannot request both an ingredient and a meal')
|
|
|
|
if ingredient is None and meal is None:
|
|
raise ValueError('Must specify either an ingredient or a meal to request')
|
|
|
|
if meal is not None and meal.id < 0:
|
|
raise ValueError('Meal must have a valid id')
|
|
|
|
if ingredient is not None and ingredient.id < 0:
|
|
await insert_ingredient(conn, ingredient)
|
|
|
|
item = ShoppingListItem(ingredient=ingredient, person=person, meal=meal)
|
|
|
|
validate_request(item)
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListItem (ingredient_id, person_id, meal_id, created_date)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (item.ingredient_id, item.person_id, item.meal_id, item.created_date.isoformat())) as cursor:
|
|
item.id = cursor.lastrowid
|
|
|
|
return item
|
|
|
|
async def remove_request(conn, person: Person, meal: Optional[Meal] = None, ingredient: Optional[Ingredient] = None) -> None:
|
|
failed = True
|
|
if meal is not None:
|
|
# Ensure a record is removed
|
|
async with conn.execute('''
|
|
DELETE FROM ShoppingListItem
|
|
WHERE list_id IS NULL AND meal_id = ?
|
|
''', (meal.id,)) as cursor:
|
|
if cursor.rowcount > 0:
|
|
failed = False
|
|
|
|
elif ingredient is not None:
|
|
# Ensure a record is removed
|
|
async with conn.execute('''
|
|
DELETE FROM ShoppingListItem
|
|
WHERE list_id IS NULL AND ingredient_id = ? AND person_id = ?
|
|
''', (ingredient.id, person.id)) as cursor:
|
|
if cursor.rowcount > 0:
|
|
failed = False
|
|
|
|
if failed:
|
|
raise ValueError('Must specify either a meal or an ingredient to remove')
|
|
|
|
async def find_items_by_list_id(conn, list_id: Optional[int]) -> AsyncIterator[ShoppingListItem]:
|
|
# Join Ingredient and Product to also load ingredient and product
|
|
ingredient_keys = [f'ingredient.{key}' for key in Ingredient.KEYS]
|
|
product_keys = [f'product.{key}' for key in Product.KEYS]
|
|
request_keys = [f'shoppinglistitem.{key}' for key in ShoppingListItem.KEYS]
|
|
person_keys = [f'person.{key}' for key in Person.KEYS]
|
|
|
|
select = f'''
|
|
SELECT {','.join(ingredient_keys + product_keys + request_keys + person_keys)}
|
|
FROM ShoppingListItem
|
|
LEFT JOIN Ingredient ON ShoppingListItem.ingredient_id = Ingredient.id
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
LEFT JOIN Person ON ShoppingListItem.person_id = Person.id
|
|
'''
|
|
|
|
where, params = ' WHERE list_id IS NULL', ()
|
|
if list_id is not None:
|
|
where, params = ' WHERE list_id = ?', (list_id,)
|
|
|
|
cursor = await conn.execute(select + where, params)
|
|
|
|
async for row in cursor:
|
|
product_keys = {k:v for k,v in zip(Product.KEYS, row[len(Ingredient.KEYS):len(Ingredient.KEYS) + len(Product.KEYS)])}
|
|
product = Product(**product_keys) if product_keys['id'] else None
|
|
|
|
ingredient_keys = {k:v for k,v in zip(Ingredient.KEYS, row[:len(Ingredient.KEYS)])}
|
|
ingredient = Ingredient(**ingredient_keys, product=product) if ingredient_keys['id'] else None
|
|
|
|
person_keys = {k:v for k,v in zip(Person.KEYS, row[-len(Person.KEYS):])}
|
|
person = Person(**person_keys) if person_keys['id'] else None
|
|
|
|
request_keys = {k:v for k,v in zip(ShoppingListItem.KEYS, row[len(Ingredient.KEYS) + len(Product.KEYS):-len(Person.KEYS)])}
|
|
request = ShoppingListItem(**request_keys, ingredient=ingredient, person=person)
|
|
|
|
if request.meal_id is not None:
|
|
request.meal = await find_meal_by_id(conn, request.meal_id)
|
|
|
|
yield request
|
|
|
|
async def load_shopping_list(conn, id: int) -> ShoppingList:
|
|
shopping_list = None
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ShoppingList.KEYS)} FROM ShoppingList
|
|
WHERE id = ?
|
|
LIMIT 1
|
|
''', (id,)) as cursor:
|
|
async for row in cursor:
|
|
shopping_list = ShoppingList(**{k:v for k,v in zip(ShoppingList.KEYS, row)})
|
|
break
|
|
|
|
if shopping_list:
|
|
async for item in find_items_by_list_id(conn, shopping_list.id):
|
|
shopping_list.items.append(item)
|
|
|
|
return shopping_list
|
|
|
|
async def get_purchased_ingredients(conn, meal_ids: List[int]) -> AsyncIterator[ShoppingListItem]:
|
|
if not meal_ids:
|
|
return
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ShoppingListItem.KEYS)}
|
|
FROM ShoppingListItem
|
|
WHERE meal_id IN ({','.join(['?'] * len(meal_ids))})
|
|
''', meal_ids) as cursor:
|
|
async for row in cursor:
|
|
yield ShoppingListItem(**{k:v for k,v in zip(ShoppingListItem.KEYS, row)}) |