250 lines
7.2 KiB
Python
250 lines
7.2 KiB
Python
import datetime
|
|
import json
|
|
from typing import Any, AsyncIterator, ClassVar, Iterable, List, Optional, Tuple, cast
|
|
|
|
from pydantic import Field
|
|
from common import ApiModel
|
|
|
|
from ingredients import Ingredient, find_ingredients_by_recipe_id
|
|
from persons import Person
|
|
|
|
|
|
class Recipe(ApiModel):
|
|
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] = Field(default_factory=list)
|
|
ingredients: List[Ingredient] = Field(default_factory=list)
|
|
based_on_recipe: Optional[int] = None
|
|
|
|
date_created: datetime.datetime = Field(
|
|
default_factory=lambda: datetime.datetime.now().astimezone()
|
|
)
|
|
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 NOT NULL,
|
|
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)
|
|
);"""
|
|
)
|
|
# Useful indexes for filtering/pagination
|
|
await conn.execute("CREATE INDEX IF NOT EXISTS idx_recipe_hidden_id ON Recipe(date_hidden, id);")
|
|
await conn.execute("CREATE INDEX IF NOT EXISTS idx_recipe_name_hidden_id ON Recipe(name, date_hidden, id);")
|
|
|
|
|
|
def _as_insert_field(recipe: Recipe, name: str):
|
|
value = getattr(recipe, name)
|
|
if name == "image_urls":
|
|
return json.dumps(value)
|
|
if isinstance(value, datetime.datetime):
|
|
return value.isoformat()
|
|
|
|
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 = ?, hidden_by_id = ?
|
|
WHERE id = ?
|
|
""",
|
|
(datetime.datetime.now().astimezone().isoformat(), person.id, recipe_id),
|
|
)
|
|
|
|
|
|
def row_to_recipe(col_tuples: Iterable[Tuple[str, object]]) -> Recipe:
|
|
d: dict[str, Any] = {k: v for k, v in col_tuples}
|
|
img_raw = (
|
|
cast(str, d["image_urls"]) if not isinstance(d["image_urls"], list) else d["image_urls"]
|
|
)
|
|
d["image_urls"] = cast(List[str], json.loads(img_raw) if isinstance(img_raw, str) else img_raw)
|
|
return Recipe(**d)
|
|
|
|
|
|
async def find_recipe_by_id(conn, recipe_id: int) -> Optional[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(list(zip(Recipe.KEYS, row)))
|
|
return None
|
|
|
|
|
|
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(list(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(list(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)
|
|
|
|
|
|
# Paged queries for v1 cursor/limit support
|
|
async def get_all_paged(conn, after_id: Optional[int], limit: int) -> AsyncIterator[Recipe]:
|
|
after = after_id if after_id is not None else -1
|
|
async with conn.execute(
|
|
f"""
|
|
SELECT {','.join(Recipe.KEYS)}
|
|
FROM Recipe
|
|
WHERE date_hidden IS NULL AND id > ?
|
|
ORDER BY id
|
|
LIMIT ?
|
|
""",
|
|
(after, limit),
|
|
) as cursor:
|
|
async for row in cursor:
|
|
yield row_to_recipe(list(zip(Recipe.KEYS, row)))
|
|
|
|
|
|
async def find_recipes_by_name_paged(
|
|
conn, name: str, after_id: Optional[int], limit: int
|
|
) -> AsyncIterator[Recipe]:
|
|
after = after_id if after_id is not None else -1
|
|
async with conn.execute(
|
|
f"""
|
|
SELECT {','.join(Recipe.KEYS)}
|
|
FROM Recipe
|
|
WHERE name LIKE ? AND date_hidden IS NULL AND id > ?
|
|
ORDER BY id
|
|
LIMIT ?
|
|
""",
|
|
(f"%{name}%", after, limit),
|
|
) as cursor:
|
|
async for row in cursor:
|
|
yield row_to_recipe(list(zip(Recipe.KEYS, row)))
|
|
|
|
|
|
async def compute_prev_cursor(
|
|
conn, first_id: int, limit: int, name: Optional[str] = None
|
|
) -> Optional[str]:
|
|
"""Compute a prevCursor string for paginated recipes.
|
|
|
|
Strategy: look up to `limit` rows before `first_id` (respecting optional name LIKE filter).
|
|
If there are at least `limit` rows, set cursor to just before the earliest id in that window.
|
|
"""
|
|
if limit <= 0:
|
|
return None
|
|
if name:
|
|
query = f"""
|
|
SELECT id
|
|
FROM Recipe
|
|
WHERE name LIKE ? AND date_hidden IS NULL AND id < ?
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
"""
|
|
params = (f"%{name}%", first_id, limit)
|
|
else:
|
|
query = f"""
|
|
SELECT id
|
|
FROM Recipe
|
|
WHERE date_hidden IS NULL AND id < ?
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
"""
|
|
params = (first_id, limit)
|
|
|
|
async with conn.execute(query, params) as c:
|
|
prev_ids = [row[0] async for row in c]
|
|
if len(prev_ids) == limit and prev_ids:
|
|
return str(min(prev_ids) - 1)
|
|
return None
|
|
|
|
|
|
async def count_all(conn) -> int:
|
|
cursor = await conn.execute(
|
|
f"""
|
|
SELECT COUNT(1)
|
|
FROM Recipe
|
|
WHERE date_hidden IS NULL
|
|
"""
|
|
)
|
|
row = await cursor.fetchone()
|
|
return int(row[0]) if row else 0
|
|
|
|
|
|
async def count_by_name(conn, name: str) -> int:
|
|
cursor = await conn.execute(
|
|
f"""
|
|
SELECT COUNT(1)
|
|
FROM Recipe
|
|
WHERE name LIKE ? AND date_hidden IS NULL
|
|
""",
|
|
(f"%{name}%",),
|
|
)
|
|
row = await cursor.fetchone()
|
|
return int(row[0]) if row else 0
|