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 List, ClassVar, Tuple, Optional

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

    date_created: Optional[datetime.datetime] = None
    created_by_id: Optional[int] = None
    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,
            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)
        );''')

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

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) -> List[Recipe]:
    async with conn.execute(f'''
        SELECT {','.join(Recipe.KEYS)} FROM Recipe
        WHERE name LIKE ?
    ''', (f'%{name}%',)) as cursor:
        async for row in cursor:
            yield row_to_recipe(zip(Recipe.KEYS, row))

async def get_all(conn) -> List[Recipe]:
    async with conn.execute(f'''
        SELECT {','.join(Recipe.KEYS)} FROM Recipe
    ''') 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)