144 lines
No EOL
5 KiB
Python
144 lines
No EOL
5 KiB
Python
from typing import List, ClassVar
|
|
from pydantic import BaseModel
|
|
from persons import Person
|
|
from ingredients import Ingredient, insert_ingredient
|
|
from products import Product
|
|
from recipes import Recipe, row_to_recipe
|
|
|
|
import datetime
|
|
|
|
class Meal(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'date']
|
|
id: int
|
|
date: datetime.datetime
|
|
chef: List[Person] = []
|
|
cleanup: List[Person] = []
|
|
consumers: List[Person] = []
|
|
recipes: List[Recipe] = []
|
|
extra_ingredients: List[Ingredient] = []
|
|
|
|
async def create(conn):
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS Meal (
|
|
id INTEGER PRIMARY KEY,
|
|
date TEXT UNIQUE
|
|
);''')
|
|
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS MealParticipant (
|
|
meal_id INTEGER,
|
|
person_id INTEGER,
|
|
role TEXT,
|
|
FOREIGN KEY(meal_id) REFERENCES Meal(id),
|
|
FOREIGN KEY(person_id) REFERENCES Person(id)
|
|
);''')
|
|
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS MealRecipe (
|
|
meal_id INTEGER,
|
|
recipe_id INTEGER,
|
|
FOREIGN KEY(meal_id) REFERENCES Meal(id),
|
|
FOREIGN KEY(recipe_id) REFERENCES Recipe(id)
|
|
);''')
|
|
|
|
async def insert_meal_participant(conn, meal_id: int, person_id: int, role: str):
|
|
await conn.execute('''
|
|
INSERT INTO MealParticipant (meal_id, person_id, role)
|
|
VALUES (?, ?, ?)
|
|
''', (meal_id, person_id, role))
|
|
|
|
async def insert_meal_recipe(conn, meal_id: int, recipe_id: int):
|
|
await conn.execute('''
|
|
INSERT INTO MealRecipe (meal_id, recipe_id)
|
|
VALUES (?, ?)
|
|
''', (meal_id, recipe_id))
|
|
|
|
async def insert_meal(conn, meal: Meal):
|
|
async with conn.execute('''
|
|
INSERT INTO Meal (date)
|
|
VALUES (?)
|
|
''', (meal.date,)) as cursor:
|
|
meal.id = cursor.lastrowid
|
|
|
|
for person in meal.chef:
|
|
await insert_meal_participant(conn, meal.id, person.id, 'chef')
|
|
|
|
for person in meal.cleanup:
|
|
await insert_meal_participant(conn, meal.id, person.id, 'cleanup')
|
|
|
|
for person in meal.consumers:
|
|
await insert_meal_participant(conn, meal.id, person.id, 'consumer')
|
|
|
|
for recipe in meal.recipes:
|
|
await insert_meal_recipe(conn, meal.id, recipe.id)
|
|
|
|
for ingredient in meal.extra_ingredients:
|
|
ingredient.meal_id = meal.id
|
|
await insert_ingredient(conn, ingredient)
|
|
|
|
async def find_meal_by_id(conn, meal_id: int) -> Meal:
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(Meal.KEYS)} FROM Meal
|
|
WHERE id = ?
|
|
LIMIT 1
|
|
''', (meal_id,)) as cursor:
|
|
async for row in cursor:
|
|
return Meal(**{k:v for k,v in zip(Meal.KEYS, row)})
|
|
|
|
async def find_meal_by_date(conn, date: datetime) -> Meal:
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(Meal.KEYS)} FROM Meal
|
|
WHERE date = ?
|
|
LIMIT 1
|
|
''', (date,)) as cursor:
|
|
async for row in cursor:
|
|
return Meal(**{k:v for k,v in zip(Meal.KEYS, row)})
|
|
|
|
async def find_meals_by_date_range(conn, start: datetime, end: datetime) -> List[Meal]:
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(Meal.KEYS)} FROM Meal
|
|
WHERE date >= ? AND date <= ?
|
|
''', (start, end)) as cursor:
|
|
result = []
|
|
async for row in cursor:
|
|
result.append(Meal(**{k:v for k,v in zip(Meal.KEYS, row)}))
|
|
return result
|
|
|
|
async def load_participants(conn, meal: Meal) -> None:
|
|
async with conn.execute(f'''
|
|
SELECT person_id, role FROM MealParticipant
|
|
WHERE meal_id = ?
|
|
''', (meal.id,)) as cursor:
|
|
async for row in cursor:
|
|
person = await Person.find_person_by_id(conn, row[0])
|
|
if row[1] == 'chef':
|
|
meal.chef.append(person)
|
|
elif row[1] == 'cleanup':
|
|
meal.cleanup.append(person)
|
|
elif row[1] == 'consumer':
|
|
meal.consumers.append(person)
|
|
else:
|
|
raise Exception(f'Unknown role: {row[1]}')
|
|
|
|
async def load_recipes(conn, meal: Meal) -> None:
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(Recipe.KEYS)} FROM Recipe
|
|
JOIN MealRecipe ON MealRecipe.recipe_id = Recipe.id
|
|
WHERE MealRecipe.meal_id = ?
|
|
''', (meal.id,)) as cursor:
|
|
async for row in cursor:
|
|
meal.recipes.append(row_to_recipe(zip(Recipe.KEYS, row)))
|
|
|
|
async def load_extra_ingredients(conn, meal: Meal) -> None:
|
|
ingredient_cols = [f'Ingredient.{k}' for k in Ingredient.KEYS]
|
|
product_cols = [f'Product.{k}' for k in Product.KEYS]
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ingredient_cols + product_cols)} FROM Ingredient
|
|
JOIN Product ON Ingredient.product_id = Product.id
|
|
WHERE meal_id = ?
|
|
''', (meal.id,)) as cursor:
|
|
async for row in cursor:
|
|
ingredient = Ingredient(**{k:v for k,v in zip(Ingredient.KEYS, row[:len(Ingredient.KEYS)])})
|
|
ingredient.product = Product(**{k:v for k,v in zip(Product.KEYS, row[len(Ingredient.KEYS):])})
|
|
meal.extra_ingredients.append(ingredient) |