munch-ease-backend/recipe/db.py

import aiosqlite

from product 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']
    id: int
    name: str
    line: str
    unit: str
    quantity: float
    preparation: str
    product_id: Optional[int] = None
    recipe_id: Optional[int] = None
    product: Product = None

class Recipe(BaseModel):
    KEYS: ClassVar[List[str]] = ['id', 'name', 'link', 'raw_data']
    id: int
    name: str
    link: str
    raw_data: str
    ingredients: List[Ingredient] = []

async def create(conn):
    await conn.execute('''
        CREATE TABLE IF NOT EXISTS Recipe (
            id INTEGER PRIMARY KEY,
            name TEXT,
            link TEXT,
            raw_data TEXT
        );''')

    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,
            FOREIGN KEY (product_id) REFERENCES Product(id),
            FOREIGN KEY (recipe_id) REFERENCES Recipe(id)
        );''')
    
async def insert_ingredient(conn, ingredient: Ingredient):
    async with conn.execute('''
        INSERT INTO Ingredient (name, line, preparation, unit, quantity, product_id, recipe_id)
        VALUES (?, ?, ?, ?, ?, ?, ?)
    ''', (ingredient.name, ingredient.line, ingredient.preparation, ingredient.unit, ingredient.quantity, ingredient.product_id, ingredient.recipe_id)) as cursor:
        ingredient.id = cursor.lastrowid

async def insert_recipe(conn, recipe: Recipe):
    async with conn.execute('''
        INSERT INTO Recipe (name, link, raw_data)
        VALUES (?, ?, ?)
    ''', (recipe.name, recipe.link, recipe.raw_data)) as cursor:
        recipe.id = cursor.lastrowid

async def find_recipe_by_id(conn, recipe_id: int) -> Recipe:
    async with conn.execute(f'''
        SELECT {','.join(Recipe.KEYS)} FROM Recipe
        WHERE id = ?
        LIMIT 1
    ''', (recipe_id,)) as cursor:
        async for row in cursor:
            return Recipe(**{k:v for k,v in zip(Recipe.KEYS, row)})
        
async def find_ingredients_by_recipe_id(conn, recipe_id: int) -> List[Ingredient]:
    async with conn.execute(f'''
        SELECT {','.join(Ingredient.KEYS)} FROM Ingredient
        WHERE recipe_id = ?
    ''', (recipe_id,)) as cursor:
        async for row in cursor:
            yield Ingredient(**{k:v for k,v in zip(Ingredient.KEYS, row)})