197 lines
6.7 KiB
Python
197 lines
6.7 KiB
Python
from typing import Any, AsyncIterator, ClassVar, List, Optional, Dict
|
|
|
|
from pydantic import field_validator, Field
|
|
from common import ApiModel
|
|
|
|
from products import Product
|
|
from units import ALL_UNITS
|
|
|
|
|
|
class Ingredient(ApiModel):
|
|
KEYS: ClassVar[List[str]] = [
|
|
"id",
|
|
"name",
|
|
"line",
|
|
"preparation",
|
|
"unit",
|
|
"quantity",
|
|
"product_id",
|
|
"recipe_id",
|
|
"meal_id",
|
|
]
|
|
id: int = -1
|
|
name: str
|
|
line: str
|
|
unit: str = Field(
|
|
title="Unit",
|
|
description="Measurement unit (enum values are advisory; runtime accepts any string)",
|
|
json_schema_extra={"enum": [u.name for u in ALL_UNITS]},
|
|
)
|
|
quantity: float | str
|
|
preparation: str
|
|
product_id: Optional[int] = None
|
|
recipe_id: Optional[int] = None
|
|
meal_id: Optional[int] = None
|
|
product: Optional[Product] = None
|
|
|
|
# Ensure quantity is stored as a float even if provided as a string in tests
|
|
@field_validator("quantity", mode="before")
|
|
@classmethod
|
|
def _coerce_quantity(cls, v: Any) -> Any:
|
|
if isinstance(v, str):
|
|
try:
|
|
return float(v)
|
|
except ValueError:
|
|
return v
|
|
return v
|
|
|
|
|
|
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)
|
|
);"""
|
|
)
|
|
# 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
|
|
|
|
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,),
|
|
)
|