munch-ease-backend/recipes/db.py

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)
        );"""
    )


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