175 lines
6.3 KiB
Python
175 lines
6.3 KiB
Python
from typing import AsyncIterator, List, Optional
|
|
|
|
from ingredients.models import Ingredient
|
|
from products.models import Product
|
|
|
|
|
|
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,
|
|
household_id INTEGER,
|
|
FOREIGN KEY (product_id) REFERENCES Product(id),
|
|
FOREIGN KEY (recipe_id) REFERENCES Recipe(id),
|
|
FOREIGN KEY (meal_id) REFERENCES Meal(id)
|
|
);"""
|
|
)
|
|
# Useful indexes
|
|
await conn.execute(
|
|
"CREATE INDEX IF NOT EXISTS idx_ingredient_recipe_id ON Ingredient(recipe_id);"
|
|
)
|
|
await conn.execute("CREATE INDEX IF NOT EXISTS idx_ingredient_meal_id ON Ingredient(meal_id);")
|
|
|
|
|
|
async def insert_ingredient(conn, ingredient: Ingredient):
|
|
if ingredient.product:
|
|
ingredient.product_id = ingredient.product.id
|
|
|
|
if ingredient.product_id is None or ingredient.product_id < 0:
|
|
ingredient.product_id = None
|
|
|
|
# Disallow empty ingredients (no product and no textual content)
|
|
name = (ingredient.name or "").strip()
|
|
line = (ingredient.line or "").strip()
|
|
has_product = ingredient.product_id is not None and ingredient.product_id >= 0
|
|
if not has_product and name == "" and line == "":
|
|
raise ValueError("Ingredient must include a name or line or a product")
|
|
|
|
# Enforce positive quantity at repository layer as well
|
|
try:
|
|
q = float(ingredient.quantity)
|
|
except Exception:
|
|
q = ingredient.quantity
|
|
if isinstance(q, (int, float)) and q <= 0:
|
|
raise ValueError("Ingredient quantity must be greater than 0")
|
|
|
|
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_ingredient_by_id(conn, ingredient_id: int) -> Optional[Ingredient]:
|
|
ingredient_cols = [f"ingredient.{key}" for key in Ingredient.KEYS]
|
|
product_cols = [f"product.{key}" for key in Product.KEYS]
|
|
|
|
async with conn.execute(
|
|
f"""
|
|
SELECT {",".join(ingredient_cols + product_cols)} FROM Ingredient
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
WHERE Ingredient.id = ?
|
|
""",
|
|
(ingredient_id,),
|
|
) as cursor:
|
|
async for row in cursor:
|
|
product_map = {k: v for k, v in zip(Product.KEYS, row[len(Ingredient.KEYS) :])}
|
|
product = Product(**product_map) if product_map["id"] else None
|
|
return Ingredient(
|
|
**{k: v for k, v in zip(Ingredient.KEYS, row[: len(Ingredient.KEYS)])},
|
|
product=product,
|
|
)
|
|
return None
|
|
|
|
|
|
async def find_ingredients_by_recipe_id(conn, recipe_id: int) -> AsyncIterator[Ingredient]:
|
|
ingredient_cols = [f"ingredient.{key}" for key in Ingredient.KEYS]
|
|
product_cols = [f"product.{key}" for key in Product.KEYS]
|
|
|
|
async with conn.execute(
|
|
f"""
|
|
SELECT {",".join(ingredient_cols + product_cols)} FROM Ingredient
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
WHERE recipe_id = ?
|
|
""",
|
|
(recipe_id,),
|
|
) as cursor:
|
|
async for row in cursor:
|
|
product_map = {k: v for k, v in zip(Product.KEYS, row[len(Ingredient.KEYS) :])}
|
|
product = Product(**product_map) if product_map["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_recipe_ids(
|
|
conn, recipe_ids: List[int]
|
|
) -> dict[int, List[Ingredient]]:
|
|
"""Fetch ingredients for many recipes in one query. Returns recipe_id -> [Ingredient]."""
|
|
if not recipe_ids:
|
|
return {}
|
|
placeholders = ",".join(["?"] * len(recipe_ids))
|
|
ingredient_cols = [f"ingredient.{key}" for key in Ingredient.KEYS]
|
|
product_cols = [f"product.{key}" for key in Product.KEYS]
|
|
query = f"""
|
|
SELECT {",".join(ingredient_cols + product_cols)}
|
|
FROM Ingredient AS ingredient
|
|
LEFT JOIN Product AS product ON ingredient.product_id = product.id
|
|
WHERE ingredient.recipe_id IN ({placeholders})
|
|
ORDER BY ingredient.recipe_id, ingredient.id
|
|
"""
|
|
result: dict[int, List[Ingredient]] = {rid: [] for rid in recipe_ids}
|
|
async with conn.execute(query, recipe_ids) as cursor:
|
|
async for row in cursor:
|
|
product_map = {k: v for k, v in zip(Product.KEYS, row[len(Ingredient.KEYS) :])}
|
|
product = Product(**product_map) if product_map["id"] else None
|
|
ing = Ingredient(
|
|
**{k: v for k, v in zip(Ingredient.KEYS, row[: len(Ingredient.KEYS)])},
|
|
product=product,
|
|
)
|
|
if ing.recipe_id is not None:
|
|
result.setdefault(int(ing.recipe_id), []).append(ing)
|
|
return result
|
|
|
|
|
|
async def find_ingredients_by_meal_id(conn, meal_id: int) -> AsyncIterator[Ingredient]:
|
|
ingredient_cols = [f"ingredient.{key}" for key in Ingredient.KEYS]
|
|
product_cols = [f"product.{key}" for key in Product.KEYS]
|
|
|
|
async with conn.execute(
|
|
f"""
|
|
SELECT {",".join(ingredient_cols + product_cols)} FROM Ingredient
|
|
LEFT JOIN Product ON Ingredient.product_id = Product.id
|
|
WHERE meal_id = ?
|
|
""",
|
|
(meal_id,),
|
|
) as cursor:
|
|
async for row in cursor:
|
|
product_map = {k: v for k, v in zip(Product.KEYS, row[len(Ingredient.KEYS) :])}
|
|
product = Product(**product_map) if product_map["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,),
|
|
)
|