400 lines
No EOL
17 KiB
Python
400 lines
No EOL
17 KiB
Python
from meals import Meal, find_meals_by_date_range, find_meal_by_id
|
|
from ingredients import Ingredient, insert_ingredient
|
|
from persons import Person
|
|
from products import Product
|
|
|
|
from pydantic import BaseModel
|
|
from typing import AsyncIterator, List, ClassVar, Optional, Tuple
|
|
|
|
from datetime import datetime, timedelta
|
|
|
|
class ShoppingListRequest(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'ingredient_id', 'list_id', 'person_id', 'meal_id', 'created_date']
|
|
id: int = -1
|
|
list_id: int
|
|
|
|
ingredient_id: Optional[int] = None
|
|
ingredient: Optional[Ingredient] = None
|
|
|
|
person_id: Optional[int] = None
|
|
person: Optional[Person] = None
|
|
|
|
meal_id: Optional[int] = None
|
|
meal: Optional[Meal] = None
|
|
|
|
created_date: datetime = datetime.now()
|
|
|
|
class ShoppingListResult(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'product_id', 'list_id', 'quantity', 'unit', 'created_date', 'found_date']
|
|
id: int = -1
|
|
list_id: int
|
|
product_id: int
|
|
product: Optional[Product] = None
|
|
|
|
quantity: float
|
|
unit: str
|
|
|
|
created_date: datetime = datetime.now()
|
|
found_date: Optional[datetime] = None
|
|
|
|
class ShoppingList(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'created_date', 'purchased_date']
|
|
id: int = -1
|
|
created_date: datetime = datetime.now()
|
|
purchased_date: Optional[datetime] = None
|
|
|
|
requests: List[ShoppingListRequest] = []
|
|
results: List[ShoppingListResult] = []
|
|
|
|
async def create(conn):
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS ShoppingList (
|
|
id INTEGER PRIMARY KEY,
|
|
created_date TEXT,
|
|
purchased_date TEXT
|
|
);''')
|
|
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS ShoppingListRequest (
|
|
id INTEGER PRIMARY KEY,
|
|
ingredient_id INTEGER,
|
|
list_id INTEGER,
|
|
person_id INTEGER,
|
|
meal_id INTEGER,
|
|
created_date TEXT,
|
|
FOREIGN KEY(ingredient_id) REFERENCES Ingredient(id),
|
|
FOREIGN KEY(person_id) REFERENCES Person(id),
|
|
FOREIGN KEY(meal_id) REFERENCES Meal(id),
|
|
FOREIGN KEY(list_id) REFERENCES ShoppingList(id)
|
|
);''')
|
|
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS ShoppingListResult (
|
|
id INTEGER PRIMARY KEY,
|
|
product_id INTEGER,
|
|
list_id INTEGER,
|
|
quantity REAL,
|
|
unit TEXT,
|
|
created_date TEXT DEFAULT CURRENT_TIMESTAMP,
|
|
found_date TEXT,
|
|
FOREIGN KEY(product_id) REFERENCES Product(id),
|
|
FOREIGN KEY(list_id) REFERENCES ShoppingList(id)
|
|
);''')
|
|
|
|
def validate_request(request: ShoppingListRequest) -> None:
|
|
# A request must always have a list id
|
|
if request.list_id < 0:
|
|
raise ValueError('Request must have a list id')
|
|
|
|
# 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')
|
|
|
|
if request.ingredient and request.meal:
|
|
raise ValueError('Request cannot have both an ingredient and a meal')
|
|
|
|
# If an ingredient is provided, it must have a person
|
|
if request.ingredient and not request.person:
|
|
raise ValueError('Ingredient requests must have a person')
|
|
|
|
async def insert_shopping_list(conn, shopping_list: ShoppingList):
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingList (created_date, purchased_date)
|
|
VALUES (CURRENT_TIMESTAMP, NULL)
|
|
''') as cursor:
|
|
shopping_list.id = cursor.lastrowid
|
|
|
|
for request in shopping_list.requests:
|
|
request.list_id = shopping_list.id
|
|
|
|
validate_request(request)
|
|
|
|
if request.ingredient and request.ingredient.id < 0:
|
|
await insert_ingredient(conn, request.ingredient)
|
|
|
|
if request.ingredient:
|
|
request.ingredient_id = request.ingredient.id
|
|
|
|
if request.meal:
|
|
request.meal_id = request.meal.id
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListRequest (ingredient_id, list_id, person_id, meal_id, created_date)
|
|
VALUES (?, ?, ?, ?, ?)
|
|
''', (request.ingredient_id, shopping_list.id, request.person_id, request.meal_id, request.created_date)) as cursor:
|
|
request.id = cursor.lastrowid
|
|
|
|
for item in shopping_list.results:
|
|
item.product_id = item.product.id
|
|
item.list_id = shopping_list.id
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListResult (product_id, list_id, quantity, unit, created_date, found_date)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
''', (item.product_id, item.list_id, item.quantity, item.unit, item.created_date, item.found_date)) as cursor:
|
|
item.id = cursor.lastrowid
|
|
|
|
async def find_request(conn, id: int) -> Optional[ShoppingListRequest]:
|
|
# Join Ingredient and Product to also load ingredient and product
|
|
product_keys = [f'product.{key}' for key in Product.KEYS]
|
|
ingredient_keys = [f'ingredient.{key}' for key in Ingredient.KEYS]
|
|
person_keys = [f'person.{key}' for key in Person.KEYS]
|
|
request_keys = [f'shoppinglistrequest.{key}' for key in ShoppingListRequest.KEYS]
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(product_keys + ingredient_keys + person_keys + request_keys)}
|
|
FROM ShoppingListRequest
|
|
LEFT JOIN Ingredient ON ShoppingListRequest.ingredient_id = Ingredient.id
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
LEFT JOIN Person ON ShoppingListRequest.person_id = Person.id
|
|
WHERE ShoppingListRequest.id = ?
|
|
''', (id,)) as cursor:
|
|
async for row in cursor:
|
|
product_keys = {k:v for k,v in zip(Product.KEYS, row[: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(Product.KEYS):len(Product.KEYS) + 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(ShoppingListRequest.KEYS, row[len(Product.KEYS) + len(Ingredient.KEYS):-len(Person.KEYS)])}
|
|
request = ShoppingListRequest(**request_keys, ingredient=ingredient, person=person)
|
|
|
|
if request.meal_id is not None:
|
|
request.meal = await find_meal_by_id(conn, request.meal_id)
|
|
|
|
return request
|
|
|
|
async def find_requests_by_list_id(conn, list_id: int) -> AsyncIterator[ShoppingListRequest]:
|
|
# 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'shoppinglistrequest.{key}' for key in ShoppingListRequest.KEYS]
|
|
person_keys = [f'person.{key}' for key in Person.KEYS]
|
|
|
|
cursor = await conn.execute(f'''
|
|
SELECT {','.join(ingredient_keys + product_keys + request_keys + person_keys)}
|
|
FROM ShoppingListRequest
|
|
LEFT JOIN Ingredient ON ShoppingListRequest.ingredient_id = Ingredient.id
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
LEFT JOIN Person ON ShoppingListRequest.person_id = Person.id
|
|
WHERE list_id = ?
|
|
''', (list_id,))
|
|
|
|
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(ShoppingListRequest.KEYS, row[len(Ingredient.KEYS) + len(Product.KEYS):-len(Person.KEYS)])}
|
|
request = ShoppingListRequest(**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 find_items_by_list_id(conn, list_id: int) -> AsyncIterator[ShoppingListResult]:
|
|
product_keys = [f'product.{key}' for key in Product.KEYS]
|
|
result_keys = [f'shoppinglistresult.{key}' for key in ShoppingListResult.KEYS]
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(product_keys + result_keys)}
|
|
FROM ShoppingListResult
|
|
LEFT JOIN Product ON ShoppingListResult.product_id = Product.id
|
|
WHERE list_id = ?
|
|
''', (list_id,)) as cursor:
|
|
async for row in cursor:
|
|
product_keys = {k:v for k,v in zip(Product.KEYS, row[:len(Product.KEYS)])}
|
|
product = Product(**product_keys) if product_keys['id'] else None
|
|
|
|
result_keys = {k:v for k,v in zip(ShoppingListResult.KEYS, row[len(Product.KEYS):])}
|
|
result = ShoppingListResult(**result_keys, product=product)
|
|
yield result
|
|
|
|
async def fill_related(conn, shopping_list: ShoppingList) -> ShoppingList:
|
|
async for request in find_requests_by_list_id(conn, shopping_list.id):
|
|
shopping_list.requests.append(request)
|
|
|
|
async for item in find_items_by_list_id(conn, shopping_list.id):
|
|
shopping_list.results.append(item)
|
|
|
|
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:
|
|
await fill_related(conn, shopping_list)
|
|
|
|
return shopping_list
|
|
|
|
async def _upcoming_meals(conn) -> AsyncIterator[Meal]:
|
|
start = datetime.now()
|
|
end = start + timedelta(days=7)
|
|
async for meal in find_meals_by_date_range(conn, start, end):
|
|
yield meal
|
|
|
|
async def current_shopping_list(conn) -> ShoppingList:
|
|
shopping_list = None
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ShoppingList.KEYS)} FROM ShoppingList
|
|
WHERE purchased_date IS NULL
|
|
LIMIT 1
|
|
''') as cursor:
|
|
async for row in cursor:
|
|
shopping_list = ShoppingList(**{k:v for k,v in zip(ShoppingList.KEYS, row)})
|
|
await fill_related(conn, shopping_list)
|
|
break
|
|
|
|
if not shopping_list:
|
|
shopping_list = ShoppingList()
|
|
|
|
async for meal in _upcoming_meals(conn):
|
|
shopping_list.requests.append(ShoppingListRequest(list_id=shopping_list.id, meal_id=meal.id, meal=meal, created_date=datetime.now(),))
|
|
|
|
await insert_shopping_list(conn, shopping_list)
|
|
|
|
return shopping_list
|
|
|
|
async def find_existing_result(conn, product: Product, shopping_list: ShoppingList) -> ShoppingListResult:
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ShoppingListResult.KEYS)} FROM ShoppingListResult
|
|
WHERE product_id = ? AND list_id = ?
|
|
LIMIT 1
|
|
''', (product.id, shopping_list.id)) as cursor:
|
|
async for row in cursor:
|
|
return ShoppingListResult(**{k:v for k,v in zip(ShoppingListResult.KEYS, row)}, product=product)
|
|
|
|
return None
|
|
|
|
async def get_persons_requests(conn, shopping_list: ShoppingList, person: Person) -> AsyncIterator[ShoppingListRequest]:
|
|
async for request in find_requests_by_list_id(conn, shopping_list.id):
|
|
if request.person_id == person.id:
|
|
yield request
|
|
|
|
async def request_ingredient(conn, shopping_list: ShoppingList, person: Person, ingredient: Ingredient) -> ShoppingListRequest:
|
|
if ingredient.id:
|
|
raise ValueError('How did you get an existing ingredient?')
|
|
|
|
await insert_ingredient(conn, ingredient)
|
|
|
|
request = ShoppingListRequest(ingredient_id=ingredient.id, ingredient=ingredient, list_id=shopping_list.id, person_id=person.id, created_date=datetime.now())
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListRequest (ingredient_id, list_id, person_id, created_date)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (request.ingredient_id, request.list_id, request.person_id, request.created_date)) as cursor:
|
|
request.id = cursor.lastrowid
|
|
|
|
shopping_list.requests.append(request)
|
|
return request
|
|
|
|
async def request_meal(conn, shopping_list: ShoppingList, person: Person, meal: Meal) -> ShoppingListRequest:
|
|
request = ShoppingListRequest(meal_id=meal.id, meal=meal, list_id=shopping_list.id, person_id=person.id, created_date=datetime.now())
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListRequest (meal_id, list_id, person_id, created_date)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (request.meal_id, request.list_id, request.person_id, request.created_date)) as cursor:
|
|
request.id = cursor.lastrowid
|
|
|
|
return request
|
|
|
|
async def delete_requests(conn, shopping_list: ShoppingList, meal: Meal) -> None:
|
|
await conn.execute('''
|
|
DELETE FROM ShoppingListRequest
|
|
WHERE list_id = ? AND meal_id = ?
|
|
''', (shopping_list.id, meal.id))
|
|
|
|
async def sync_persons_requested_ingredients(conn, shopping_list: ShoppingList, person: Person, requests: List[Ingredient]) -> AsyncIterator[ShoppingListRequest]:
|
|
# Delete existing and insert all as new
|
|
await conn.execute('''
|
|
DELETE FROM ShoppingListRequest
|
|
WHERE list_id = ? AND person_id = ? AND ingredient_id IS NOT NULL
|
|
''', (shopping_list.id, person.id))
|
|
|
|
for ingredient in requests:
|
|
ingredient.id = 0
|
|
yield await request_ingredient(conn, shopping_list, person, ingredient)
|
|
|
|
async def find_existing_result(conn, shopping_list: ShoppingList, product: Product, unit: str) -> Optional[ShoppingListResult]:
|
|
product_keys = [f'product.{key}' for key in Product.KEYS]
|
|
result_keys = [f'shoppinglistresult.{key}' for key in ShoppingListResult.KEYS]
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(product_keys + result_keys)}
|
|
FROM ShoppingListResult
|
|
LEFT JOIN Product ON ShoppingListResult.product_id = Product.id
|
|
WHERE ShoppingListResult.list_id = ? AND ShoppingListResult.product_id = ? AND ShoppingListResult.unit = ?
|
|
LIMIT 1
|
|
''', (shopping_list.id, product.id, unit)) as cursor:
|
|
async for row in cursor:
|
|
product_keys = {k:v for k,v in zip(Product.KEYS, row[:len(Product.KEYS)])}
|
|
product = Product(**product_keys) if product_keys['id'] else None
|
|
|
|
result_keys = {k:v for k,v in zip(ShoppingListResult.KEYS, row[len(Product.KEYS):])}
|
|
return ShoppingListResult(**result_keys, product=product)
|
|
|
|
return None
|
|
|
|
async def mark_found(conn, ingredient: Ingredient, date_found: datetime) -> Tuple[ShoppingListResult, ShoppingListResult]:
|
|
product, quantity, unit = ingredient.product, ingredient.quantity, ingredient.unit
|
|
shopping_list = await current_shopping_list(conn)
|
|
|
|
created = ShoppingListResult(product=product, product_id=product.id, list_id=shopping_list.id, quantity=quantity, unit=unit, found_date=date_found)
|
|
removed = await find_existing_result(conn, shopping_list, product, unit)
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO ShoppingListResult (product_id, list_id, quantity, unit, created_date, found_date)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
''', (created.product_id, created.list_id, quantity, unit, created.created_date, created.found_date)) as cursor:
|
|
created.id = cursor.lastrowid
|
|
|
|
if removed:
|
|
shopping_list.results = [r for r in shopping_list.results if r.id != removed.id]
|
|
await conn.execute('''
|
|
DELETE FROM ShoppingListResult
|
|
WHERE id = ?
|
|
''', (removed.id,))
|
|
|
|
return removed, created
|
|
|
|
async def unmark_found(conn, product_id: int) -> List[ShoppingListRequest]:
|
|
shopping_list = await current_shopping_list(conn)
|
|
|
|
requests = []
|
|
async with conn.execute('''
|
|
DELETE FROM ShoppingListResult
|
|
WHERE list_id = ? AND product_id = ?
|
|
''', (shopping_list.id, product_id)) as cursor:
|
|
async for row in cursor:
|
|
requests.append(row)
|
|
|
|
deleted = [r for r in shopping_list.results if r.product_id == product_id]
|
|
shopping_list.results = [r for r in shopping_list.results if r.product_id != product_id]
|
|
return deleted
|
|
|
|
async def mark_purchased(conn) -> ShoppingList:
|
|
shopping_list = await current_shopping_list(conn)
|
|
shopping_list.purchased_date = datetime.now()
|
|
|
|
await conn.execute('''
|
|
UPDATE ShoppingList
|
|
SET purchased_date = ?
|
|
WHERE id = ?
|
|
''', (shopping_list.purchased_date, shopping_list.id))
|
|
|
|
return shopping_list |