munch-ease-backend/ingredients/db.py

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