munch-ease-backend/recipes/db.py

import json, datetime

from persons import Person
from ingredients import Ingredient, find_ingredients_by_recipe_id

from pydantic import BaseModel
from typing import AsyncIterator, List, ClassVar, Tuple, Optional

class Recipe(BaseModel):
    KEYS: ClassVar[List[str]] = ['id', 'name', 'link', 'serves', 'image_urls', 'based_on_recipe', 'created_by_id', 'date_created', 'hidden_by_id', 'date_hidden']
    NON_INSERT_KEYS: ClassVar[List[str]] = ['id', 'created_date', 'hidden_by_id', 'date_hidden']
                                               
    id: int = -1
    name: str
    link: str
    serves: int
    image_urls: List[str] = []
    ingredients: List[Ingredient] = []
    based_on_recipe: Optional[int] = None

    date_created: datetime.datetime = datetime.datetime.now()
    created_by_id: Optional[int]
    created_by: Optional[Person] = None

    date_hidden: Optional[datetime.datetime] = None
    hidden_by_id: Optional[int] = None
    hidden_by: Optional[Person] = None

async def create(conn):
    await conn.execute('''
        CREATE TABLE IF NOT EXISTS Recipe (
            id INTEGER PRIMARY KEY,
            name TEXT NOT NULL,
            link TEXT NOT NULL,
            serves INTEGER NOT NULL,
            image_urls TEXT NOT NULL,
            based_on_recipe INTEGER NULL,
                       
            date_created DATETIME DEFAULT CURRENT_TIMESTAMP,
            created_by_id INTEGER NOT NULL,
                       
            date_hidden DATETIME DEFAULT NULL,
            hidden_by_id INTEGER DEFAULT NULL,
            
            FOREIGN KEY (based_on_recipe) REFERENCES Recipe(id)
            FOREIGN KEY (created_by_id) REFERENCES Person(id)
            FOREIGN KEY (hidden_by_id) REFERENCES Person(id)
        );''')

def _as_insert_field(recipe: Recipe, name: str):
    value = getattr(recipe, name)
    if name == 'image_urls':
        return json.dumps(value)
    return value

async def insert_recipe(conn, recipe: Recipe):
    fields_to_insert = [k for k in Recipe.KEYS if k not in Recipe.NON_INSERT_KEYS]
    actual_values = [_as_insert_field(recipe, k) for k in fields_to_insert]

    insert_stmt = f'''
        INSERT INTO Recipe ({','.join(fields_to_insert)})
        VALUES ({','.join(['?'] * len(fields_to_insert))})
    '''

    async with conn.execute(insert_stmt, actual_values) as cursor:
        recipe.id = cursor.lastrowid

async def hide_recipe(conn, recipe_id: int, person: Person):
    await conn.execute('''
        UPDATE Recipe
        SET date_hidden = CURRENT_TIMESTAMP, hidden_by_id = ?
        WHERE id = ?
    ''', (person.id, recipe_id))

def row_to_recipe(col_tuples: List[Tuple[str, ...]]) -> Recipe:
    d = {k:v for k,v in col_tuples}
    d['image_urls'] = json.loads(d['image_urls'])
    return Recipe(**d)

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 row_to_recipe(zip(Recipe.KEYS, row))

async def find_recipes_by_name(conn, name: str) -> AsyncIterator[Recipe]:
    async with conn.execute(f'''
        SELECT {','.join(Recipe.KEYS)} FROM Recipe
        WHERE name LIKE ? AND date_hidden IS NULL
    ''', (f'%{name}%',)) as cursor:
        async for row in cursor:
            yield row_to_recipe(zip(Recipe.KEYS, row))

async def get_all(conn) -> AsyncIterator[Recipe]:
    async with conn.execute(f'''
        SELECT {','.join(Recipe.KEYS)} FROM Recipe WHERE date_hidden IS NULL
    ''') as cursor:
        async for row in cursor:
            yield row_to_recipe(zip(Recipe.KEYS, row))

async def load_recipe_ingredients(conn, recipe: Recipe) -> None:
    async for ingredient in find_ingredients_by_recipe_id(conn, recipe.id):
        recipe.ingredients.append(ingredient)