327 lines
9.7 KiB
Python
327 lines
9.7 KiB
Python
import datetime
|
|
import json
|
|
from typing import Any, AsyncIterator, Iterable, List, Optional, Tuple, cast
|
|
|
|
from ingredients import find_ingredients_by_recipe_id
|
|
from persons.models import Person
|
|
from recipes.models import Recipe
|
|
|
|
|
|
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
|
|
|
|
|
|
# V2 scoped helpers (preserve v1 signatures)
|
|
async def insert_recipe_scoped(conn, recipe: Recipe, household_id: int):
|
|
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)}, household_id)
|
|
VALUES ({",".join(["?"] * len(fields_to_insert))}, ?)
|
|
"""
|
|
async with conn.execute(insert_stmt, (*actual_values, household_id)) 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),
|
|
)
|
|
|
|
|
|
async def hide_recipe_scoped(conn, recipe_id: int, household_id: int) -> bool:
|
|
"""Soft-delete a recipe by household for v2.
|
|
|
|
Returns True if updated, False if not found or not in household.
|
|
"""
|
|
async with conn.execute(
|
|
"""
|
|
UPDATE Recipe
|
|
SET date_hidden = ?
|
|
WHERE id = ? AND household_id = ?
|
|
""",
|
|
(datetime.datetime.now().astimezone().isoformat(), recipe_id, household_id),
|
|
) as cur:
|
|
return cur.rowcount > 0
|
|
|
|
|
|
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_recipe_by_id_scoped(conn, recipe_id: int, household_id: int) -> Optional[Recipe]:
|
|
async with conn.execute(
|
|
f"""
|
|
SELECT {",".join(Recipe.KEYS)} FROM Recipe
|
|
WHERE id = ? AND household_id = ? AND date_hidden IS NULL
|
|
LIMIT 1
|
|
""",
|
|
(recipe_id, household_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 get_all_paged_scoped(
|
|
conn, after_id: Optional[int], limit: int, household_id: 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 > ? AND household_id = ?
|
|
ORDER BY id
|
|
LIMIT ?
|
|
""",
|
|
(after, household_id, limit),
|
|
) 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 find_recipes_by_name_paged_scoped(
|
|
conn, name: str, after_id: Optional[int], limit: int, household_id: 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 > ? AND household_id = ?
|
|
ORDER BY id
|
|
LIMIT ?
|
|
""",
|
|
(f"%{name}%", after, household_id, 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 = """
|
|
SELECT id
|
|
FROM Recipe
|
|
WHERE name LIKE ? AND date_hidden IS NULL AND id < ?
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
"""
|
|
from typing import Any
|
|
|
|
params: tuple[Any, ...] = (f"%{name}%", first_id, limit)
|
|
else:
|
|
query = """
|
|
SELECT id
|
|
FROM Recipe
|
|
WHERE date_hidden IS NULL AND id < ?
|
|
ORDER BY id DESC
|
|
LIMIT ?
|
|
"""
|
|
from typing import Any
|
|
|
|
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(
|
|
"""
|
|
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_all_scoped(conn, household_id: int) -> int:
|
|
cursor = await conn.execute(
|
|
"""
|
|
SELECT COUNT(1)
|
|
FROM Recipe
|
|
WHERE date_hidden IS NULL AND household_id = ?
|
|
""",
|
|
(household_id,),
|
|
)
|
|
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(
|
|
"""
|
|
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
|
|
|
|
|
|
async def count_by_name_scoped(conn, name: str, household_id: int) -> int:
|
|
cursor = await conn.execute(
|
|
"""
|
|
SELECT COUNT(1)
|
|
FROM Recipe
|
|
WHERE name LIKE ? AND date_hidden IS NULL AND household_id = ?
|
|
""",
|
|
(f"%{name}%", household_id),
|
|
)
|
|
row = await cursor.fetchone()
|
|
return int(row[0]) if row else 0
|