Clarium/backend/books.py

"""Books slice: CRUD, progress, search/filter, and the genre list."""

import re
from collections.abc import Callable
from datetime import datetime
from typing import TypedDict, cast

import psycopg2.errors
from flask import Blueprint, Response, g, jsonify, request
from flask.typing import ResponseReturnValue

from db import Params, Row, query, query_one, query_row
from validation import ApiError, JsonObject, integer, json_body, text

bp = Blueprint("books", __name__, url_prefix="/api")

MAX_TEXT = 500
MAX_PAGES = 100_000
MAX_GENRE_ID = 32_767
GENRE_ID_PATTERN = re.compile(r"[0-9]{1,5}")
REQUIRED_FIELDS = ("title", "author", "genre_id", "total_pages")
FIELDS = set(REQUIRED_FIELDS) | {"current_page"}
VALIDATORS: dict[str, Callable[[JsonObject], str | int]] = {
    "title": lambda body: text(body, "title", MAX_TEXT),
    "author": lambda body: text(body, "author", MAX_TEXT),
    "genre_id": lambda body: integer(body, "genre_id", 1, MAX_GENRE_ID),
    "total_pages": lambda body: integer(body, "total_pages", 1, MAX_PAGES),
    "current_page": lambda body: integer(body, "current_page", 0, MAX_PAGES),
}


class BookFields(TypedDict):
    title: str
    author: str
    genre_id: int
    total_pages: int
    current_page: int


class BookRow(BookFields):
    id: int
    genre_name: str
    created_at: datetime
    updated_at: datetime


BOOK_SELECT = """
    SELECT b.id, b.title, b.author, b.genre_id, gn.name AS genre_name,
           b.total_pages, b.current_page, b.created_at, b.updated_at
      FROM books b
      JOIN genres gn ON gn.id = b.genre_id
"""


def reading_status(current_page: int, total_pages: int) -> str:
    if current_page == 0:
        return "not_started"
    if current_page == total_pages:
        return "finished"
    return "reading"


def to_json(row: BookRow) -> dict[str, object]:
    return {
        "id": row["id"],
        "title": row["title"],
        "author": row["author"],
        "genre": {"id": row["genre_id"], "name": row["genre_name"]},
        "total_pages": row["total_pages"],
        "current_page": row["current_page"],
        "status": reading_status(row["current_page"], row["total_pages"]),
        "created_at": row["created_at"].isoformat(),
        "updated_at": row["updated_at"].isoformat(),
    }


def escape_like(term: str) -> str:
    return term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")


def owned_book(book_id: int, for_update: bool = False) -> BookRow:
    row = query_one(
        BOOK_SELECT
        + " WHERE b.id = %s AND b.user_id = %s"
        + (" FOR UPDATE OF b" if for_update else ""),
        (book_id, g.user_id),
    )
    if row is None:
        raise ApiError(404, "Book not found")
    return cast(BookRow, row)


def _check_progress(book: BookFields, field: str) -> None:
    if book["current_page"] > book["total_pages"]:
        raise ApiError(
            400,
            f"current_page ({book['current_page']}) exceeds total_pages ({book['total_pages']}); "
            "lower current_page first or choose a larger total_pages",
            field,
        )


def _save(sql: str, params: Params) -> Row:
    try:
        return query_row(sql, params)
    except psycopg2.errors.ForeignKeyViolation:
        raise ApiError(
            400, "Unknown genre_id; see GET /api/genres", "genre_id"
        ) from None


@bp.get("/genres")
def list_genres() -> Response:
    return jsonify(query("SELECT id, name FROM genres ORDER BY name"))


@bp.get("/books")
def list_books() -> Response:
    sql = BOOK_SELECT + " WHERE b.user_id = %s"
    params: list[object] = [g.user_id]
    search = request.args.get("q", "").strip()
    if search:
        if len(search) > MAX_TEXT or "\x00" in search:
            raise ApiError(
                400, f"q must be plain text of at most {MAX_TEXT} characters", "q"
            )
        pattern = f"%{escape_like(search)}%"
        sql += " AND (b.title ILIKE %s OR b.author ILIKE %s)"
        params += [pattern, pattern]
    genre_id = request.args.get("genre_id", "")
    if genre_id:
        if not GENRE_ID_PATTERN.fullmatch(genre_id) or int(genre_id) > MAX_GENRE_ID:
            raise ApiError(
                400, "genre_id must be an id from GET /api/genres", "genre_id"
            )
        sql += " AND b.genre_id = %s"
        params.append(int(genre_id))
    sql += " ORDER BY b.updated_at DESC, b.id DESC"
    return jsonify([to_json(cast(BookRow, row)) for row in query(sql, tuple(params))])


@bp.post("/books")
def create_book() -> ResponseReturnValue:
    body = json_body(FIELDS)
    book = cast(
        BookFields,
        {field: VALIDATORS[field](body) for field in REQUIRED_FIELDS}
        | {
            "current_page": (
                VALIDATORS["current_page"](body) if "current_page" in body else 0
            )
        },
    )
    _check_progress(book, "current_page")
    row = _save(
        """
        INSERT INTO books (user_id, title, author, genre_id, total_pages, current_page)
        VALUES (%s, %s, %s, %s, %s, %s)
        RETURNING id
        """,
        (
            g.user_id,
            book["title"],
            book["author"],
            book["genre_id"],
            book["total_pages"],
            book["current_page"],
        ),
    )
    return jsonify(to_json(owned_book(row["id"]))), 201


@bp.get("/books/<int(max=2147483647):book_id>")
def get_book(book_id: int) -> Response:
    return jsonify(to_json(owned_book(book_id)))


@bp.patch("/books/<int(max=2147483647):book_id>")
def update_book(book_id: int) -> Response:
    body = json_body(FIELDS)
    if not body:
        raise ApiError(400, f"Provide at least one of: {', '.join(sorted(FIELDS))}")
    current: Row = dict(owned_book(book_id, for_update=True))
    book = cast(
        BookFields,
        {field: current[field] for field in FIELDS}
        | {field: VALIDATORS[field](body) for field in body},
    )
    _check_progress(book, "total_pages" if "total_pages" in body else "current_page")
    _save(
        """
        UPDATE books
           SET title = %s, author = %s, genre_id = %s, total_pages = %s, current_page = %s, updated_at = now()
         WHERE id = %s AND user_id = %s
        RETURNING id
        """,
        (
            book["title"],
            book["author"],
            book["genre_id"],
            book["total_pages"],
            book["current_page"],
            book_id,
            g.user_id,
        ),
    )
    return jsonify(to_json(owned_book(book_id)))


@bp.delete("/books/<int(max=2147483647):book_id>")
def delete_book(book_id: int) -> ResponseReturnValue:
    if (
        query_one(
            "DELETE FROM books WHERE id = %s AND user_id = %s RETURNING id",
            (book_id, g.user_id),
        )
        is None
    ):
        raise ApiError(404, "Book not found")
    return "", 204