munch-ease-backend/ingredients/repository.py

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,
            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,),
    )