82 lines
No EOL
3.3 KiB
Python
82 lines
No EOL
3.3 KiB
Python
from products import Product
|
|
|
|
from pydantic import BaseModel
|
|
from typing import List, ClassVar, Optional
|
|
|
|
class Ingredient(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'name', 'line', 'preparation', 'unit', 'quantity', 'product_id', 'recipe_id', 'meal_id']
|
|
id: int
|
|
name: str
|
|
line: str
|
|
unit: str
|
|
quantity: float
|
|
preparation: str
|
|
product_id: Optional[int] = None
|
|
recipe_id: Optional[int] = None
|
|
meal_id: Optional[int] = None
|
|
product: Optional[Product] = None
|
|
|
|
async def create(conn):
|
|
await conn.execute('''
|
|
CREATE TABLE IF NOT EXISTS Ingredient (
|
|
id INTEGER PRIMARY KEY,
|
|
name TEXT,
|
|
line TEXT,
|
|
preparation TEXT,
|
|
unit TEXT,
|
|
quantity REAL,
|
|
product_id INTEGER,
|
|
recipe_id INTEGER,
|
|
meal_id INTEGER,
|
|
FOREIGN KEY (product_id) REFERENCES Product(id),
|
|
FOREIGN KEY (recipe_id) REFERENCES Recipe(id),
|
|
FOREIGN KEY (meal_id) REFERENCES Meal(id)
|
|
);''')
|
|
|
|
|
|
async def insert_ingredient(conn, ingredient: Ingredient):
|
|
if not ingredient.product_id and ingredient.product:
|
|
ingredient.product_id = ingredient.product.id
|
|
|
|
if not ingredient.product_id:
|
|
raise ValueError('Product must be inserted before ingredient')
|
|
|
|
async with conn.execute('''
|
|
INSERT INTO Ingredient (name, line, preparation, unit, quantity, product_id, recipe_id, meal_id)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
|
|
''', (ingredient.name, ingredient.line, ingredient.preparation, ingredient.unit, ingredient.quantity, ingredient.product_id, ingredient.recipe_id, ingredient.meal_id)) as cursor:
|
|
ingredient.id = cursor.lastrowid
|
|
|
|
async def find_ingredients_by_recipe_id(conn, recipe_id: int) -> List[Ingredient]:
|
|
ingredient_keys = [f'ingredient.{key}' for key in Ingredient.KEYS]
|
|
product_keys = [f'product.{key}' for key in Product.KEYS]
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ingredient_keys + product_keys)} FROM Ingredient
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
WHERE recipe_id = ?
|
|
''', (recipe_id,)) as cursor:
|
|
async for row in cursor:
|
|
product_keys = {k:v for k,v in zip(Product.KEYS, row[len(Ingredient.KEYS):])}
|
|
product = Product(**product_keys) if product_keys['id'] else None
|
|
yield Ingredient(**{k:v for k,v in zip(Ingredient.KEYS, row[:len(Ingredient.KEYS)])}, product=product)
|
|
|
|
async def find_ingredients_by_meal_id(conn, meal_id: int) -> List[Ingredient]:
|
|
ingredient_keys = [f'ingredient.{key}' for key in Ingredient.KEYS]
|
|
product_keys = [f'product.{key}' for key in Product.KEYS]
|
|
|
|
async with conn.execute(f'''
|
|
SELECT {','.join(ingredient_keys + product_keys)} FROM Ingredient
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
WHERE meal_id = ?
|
|
''', (meal_id,)) as cursor:
|
|
async for row in cursor:
|
|
product_keys = {k:v for k,v in zip(Product.KEYS, row[len(Ingredient.KEYS):])}
|
|
product = Product(**product_keys) if product_keys['id'] else None
|
|
yield Ingredient(**{k:v for k,v in zip(Ingredient.KEYS, row[:len(Ingredient.KEYS)])}, product=product)
|
|
|
|
async def delete_ingredients_by_meal_id(conn, meal_id: int):
|
|
await conn.execute('''
|
|
DELETE FROM Ingredient
|
|
WHERE meal_id = ?
|
|
''', (meal_id,)) |