feat: add per-user books CRUD with progress and genres
Claude-Session: https://claude.ai/code/session_01M9MLit5Ko3X4s7rzC5Kv7X
This commit is contained in:
+162
-1
@@ -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/<int(max=2147483647):book_id>")
|
||||
@login_required
|
||||
def get_book(book_id: int):
|
||||
return jsonify(to_json(owned_book(book_id)))
|
||||
|
||||
|
||||
@bp.patch("/books/<int(max=2147483647):book_id>")
|
||||
@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/<int(max=2147483647):book_id>")
|
||||
@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
|
||||
|
||||
@@ -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")
|
||||
Reference in New Issue
Block a user