61 lines
No EOL
1.8 KiB
Python
61 lines
No EOL
1.8 KiB
Python
import json
|
|
|
|
from ingredients import Ingredient
|
|
|
|
from pydantic import BaseModel
|
|
from typing import List, ClassVar
|
|
|
|
class Recipe(BaseModel):
|
|
KEYS: ClassVar[List[str]] = ['id', 'name', 'link', 'image_urls', 'raw_data']
|
|
id: int
|
|
name: str
|
|
link: str
|
|
raw_data: str
|
|
image_urls: List[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,
|
|
image_urls TEXT,
|
|
raw_data TEXT
|
|
);''')
|
|
|
|
async def insert_recipe(conn, recipe: Recipe):
|
|
async with conn.execute('''
|
|
INSERT INTO Recipe (name, link, raw_data, image_urls)
|
|
VALUES (?, ?, ?, ?)
|
|
''', (recipe.name, recipe.link, recipe.raw_data, json.dumps(recipe.image_urls))) as cursor:
|
|
recipe.id = cursor.lastrowid
|
|
|
|
def row_to_recipe(row) -> Recipe:
|
|
d = {k:v for k,v in zip(Recipe.KEYS, row)}
|
|
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(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(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(row) |