From 6447251914d9b8664c7ca347b04dd443408ddb7e Mon Sep 17 00:00:00 2001 From: Malcolm Roberts Date: Fri, 2 Oct 2026 14:17:46 -0500 Subject: [PATCH] feat: add per-user books CRUD with progress and genres Claude-Session: https://claude.ai/code/session_01M9MLit5Ko3X4s7rzC5Kv7X --- backend/books.py | 163 +++++++++++++++++++++++++++++++++++- backend/tests/test_books.py | 114 +++++++++++++++++++++++++ 2 files changed, 276 insertions(+), 1 deletion(-) create mode 100644 backend/tests/test_books.py diff --git a/backend/books.py b/backend/books.py index c8cf197..bd82de6 100644 --- a/backend/books.py +++ b/backend/books.py @@ -1,4 +1,165 @@ """Books slice: CRUD, progress, search/filter, and the genre list.""" -from flask import Blueprint +import re + +import psycopg2.errors +from flask import Blueprint, g, jsonify, request + +from auth import login_required +from db import query, query_one +from validation import ApiError, integer, json_body, text bp = Blueprint("books", __name__, url_prefix="/api") + +MAX_TEXT = 500 +MAX_PAGES = 100_000 +MAX_GENRE_ID = 32_767 # SMALLINT +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 = { + "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), +} + +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: dict) -> dict: + 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: + # Backslash is Postgres's default LIKE escape character. + return term.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") + + +def owned_book(book_id: int) -> dict: + row = query_one(BOOK_SELECT + " WHERE b.id = %s AND b.user_id = %s", (book_id, g.user_id)) + if row is None: + raise ApiError(404, "Book not found") + return row + + +def _check_progress(book: dict, 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: tuple) -> dict | None: + try: + return query_one(sql, params) + except psycopg2.errors.ForeignKeyViolation: + raise ApiError(400, "Unknown genre_id; see GET /api/genres", "genre_id") from None + + +@bp.get("/genres") +@login_required +def list_genres(): + return jsonify(query("SELECT id, name FROM genres ORDER BY name")) + + +@bp.get("/books") +@login_required +def list_books(): + sql = BOOK_SELECT + " WHERE b.user_id = %s" + params: list = [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(row) for row in query(sql, tuple(params))]) + + +@bp.post("/books") +@login_required +def create_book(): + body = json_body(FIELDS) + book = {field: VALIDATORS[field](body) for field in REQUIRED_FIELDS} + book["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 + + +# int(max=...) keeps ids within Postgres INT range: larger values 404 instead of erroring in SQL. +@bp.get("/books/") +@login_required +def get_book(book_id: int): + return jsonify(to_json(owned_book(book_id))) + + +@bp.patch("/books/") +@login_required +def update_book(book_id: int): + body = json_body(FIELDS) + if not body: + raise ApiError(400, f"Provide at least one of: {', '.join(sorted(FIELDS))}") + current = owned_book(book_id) + book = {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 + """, + (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/") +@login_required +def delete_book(book_id: int): + 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 diff --git a/backend/tests/test_books.py b/backend/tests/test_books.py new file mode 100644 index 0000000..5d892c7 --- /dev/null +++ b/backend/tests/test_books.py @@ -0,0 +1,114 @@ +from tests.support import ApiTestCase + + +class BookTests(ApiTestCase): + def setUp(self): + super().setUp() + self.register() + + def test_create_returns_full_shape(self): + book = self.create_book() + self.assertEqual(book["title"], "Dune") + self.assertEqual(book["genre"], {"id": self.genre_id("Science Fiction"), "name": "Science Fiction"}) + self.assertEqual((book["current_page"], book["total_pages"], book["status"]), (0, 412, "not_started")) + self.assertIn("T", book["created_at"]) # ISO 8601 + + def test_create_trims_text(self): + book = self.create_book(title=" Dune ") + self.assertEqual(book["title"], "Dune") + + def test_progress_updates_status(self): + book = self.create_book() + reading = self.call("PATCH", f"/api/books/{book['id']}", {"current_page": 100}).get_json() + self.assertEqual((reading["current_page"], reading["status"]), (100, "reading")) + finished = self.call("PATCH", f"/api/books/{book['id']}", {"current_page": 412}).get_json() + self.assertEqual(finished["status"], "finished") + self.assertEqual(self.call("GET", f"/api/books/{book['id']}").get_json()["current_page"], 412) + + def test_edit_details(self): + book = self.create_book() + response = self.call( + "PATCH", f"/api/books/{book['id']}", + {"title": "Dune Messiah", "genre_id": self.genre_id("Fantasy"), "total_pages": 256}, + ) + self.assertEqual(response.status_code, 200) + updated = response.get_json() + self.assertEqual((updated["title"], updated["genre"]["name"], updated["total_pages"]), ("Dune Messiah", "Fantasy", 256)) + self.assertGreater(updated["updated_at"], book["updated_at"]) + + def test_create_validation(self): + valid = {"title": "Dune", "author": "Frank Herbert", "genre_id": self.genre_id("Fiction"), "total_pages": 412} + cases = [ + ({**valid, "title": " "}, "title"), + ({k: v for k, v in valid.items() if k != "author"}, "author"), + ({**valid, "title": "x" * 501}, "title"), + ({**valid, "title": "a\x00b"}, "title"), + ({**valid, "title": "\ud800"}, "title"), + ({**valid, "genre_id": 999}, "genre_id"), + ({**valid, "genre_id": "5"}, "genre_id"), + ({**valid, "total_pages": True}, "total_pages"), + ({**valid, "total_pages": 12.5}, "total_pages"), + ({**valid, "total_pages": 0}, "total_pages"), + ({**valid, "current_page": 413, "total_pages": 412}, "current_page"), + ({**valid, "user_id": 2}, "user_id"), + ] + for payload, field in cases: + with self.subTest(field=field, payload=payload): + response = self.call("POST", "/api/books", payload) + self.assertEqual(response.status_code, 400, response.get_json()) + self.assertEqual(response.get_json()["field"], field) + + def test_lowering_total_below_current_page_is_rejected_with_fix(self): + book = self.create_book(current_page=312, total_pages=400) + response = self.call("PATCH", f"/api/books/{book['id']}", {"total_pages": 300}) + self.assertEqual(response.status_code, 400) + body = response.get_json() + self.assertEqual(body["field"], "total_pages") + self.assertIn("312", body["error"]) + self.assertIn("300", body["error"]) + + def test_empty_patch_is_rejected(self): + book = self.create_book() + self.assertEqual(self.call("PATCH", f"/api/books/{book['id']}", {}).status_code, 400) + + def test_delete(self): + book = self.create_book() + self.assertEqual(self.call("DELETE", f"/api/books/{book['id']}").status_code, 204) + self.assertEqual(self.call("GET", f"/api/books/{book['id']}").status_code, 404) + self.assertEqual(self.call("DELETE", f"/api/books/{book['id']}").status_code, 404) + + def test_out_of_range_id_is_404(self): + self.assertEqual(self.call("GET", "/api/books/99999999999").status_code, 404) + + def test_list_is_most_recently_updated_first(self): + first = self.create_book(title="First") + self.create_book(title="Second") + self.call("PATCH", f"/api/books/{first['id']}", {"current_page": 1}) + titles = [b["title"] for b in self.call("GET", "/api/books").get_json()] + self.assertEqual(titles, ["First", "Second"]) + + def test_genres_sorted_and_require_login(self): + genres = self.call("GET", "/api/genres").get_json() + self.assertEqual(len(genres), 16) + self.assertEqual([g["name"] for g in genres], sorted(g["name"] for g in genres)) + self.call("POST", "/api/auth/logout") + self.assertEqual(self.call("GET", "/api/genres").status_code, 401) + + def test_books_require_login(self): + self.call("POST", "/api/auth/logout") + self.assertEqual(self.call("GET", "/api/books").status_code, 401) + + +class BookIsolationTests(ApiTestCase): + def test_other_user_cannot_see_or_modify_book(self): + self.register("alice") + book = self.create_book() + bob = self.other_user("bob") + path = f"/api/books/{book['id']}" + for method, payload in (("GET", None), ("PATCH", {"title": "Hacked"}), ("DELETE", None)): + with self.subTest(method=method): + response = self.call(method, path, payload, client=bob) + self.assertEqual(response.status_code, 404) + self.assertEqual(response.get_json(), {"error": "Book not found"}) + self.assertEqual(self.call("GET", "/api/books", client=bob).get_json(), []) + self.assertEqual(self.call("GET", path).get_json()["title"], "Dune")