mroberts
7f5d034a1f
- Require a session on every route; public routes opt out with @allow_anonymous - Split password hashing and pepper loading into passwords.py - Add a system/light/dark theme toggle backed by light-dark() colors - Ignore stale 401s from an earlier session, PATCH only changed book fields, and block overlapping journal-entry saves - Add bin/start and CoderPad Vite server settings for the pad's start/restart - Rewrite README as a mise onboarding guide; expand .gitignore - Include review-round fixes and tests
95 lines
2.7 KiB
Python
95 lines
2.7 KiB
Python
"""Notes slice: a per-book reading journal. Ownership is always derived through books.user_id."""
|
|
|
|
from datetime import datetime
|
|
from typing import TypedDict, cast
|
|
|
|
from flask import Blueprint, Response, g, jsonify
|
|
from flask.typing import ResponseReturnValue
|
|
|
|
from db import Row, query, query_one, query_row
|
|
from validation import ApiError, json_body, text
|
|
|
|
bp = Blueprint("notes", __name__, url_prefix="/api")
|
|
|
|
MAX_BODY = 10_000
|
|
NOTE_COLUMNS = "n.id, n.book_id, n.body, n.created_at, n.updated_at"
|
|
|
|
|
|
class NoteRow(TypedDict):
|
|
id: int
|
|
book_id: int
|
|
body: str
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
|
|
|
|
def to_json(row: Row) -> dict[str, object]:
|
|
note = cast(NoteRow, row)
|
|
return {
|
|
"id": note["id"],
|
|
"book_id": note["book_id"],
|
|
"body": note["body"],
|
|
"created_at": note["created_at"].isoformat(),
|
|
"updated_at": note["updated_at"].isoformat(),
|
|
}
|
|
|
|
|
|
def _require_book(book_id: int) -> None:
|
|
if (
|
|
query_one(
|
|
"SELECT 1 FROM books WHERE id = %s AND user_id = %s", (book_id, g.user_id)
|
|
)
|
|
is None
|
|
):
|
|
raise ApiError(404, "Book not found")
|
|
|
|
|
|
@bp.get("/books/<int(max=2147483647):book_id>/notes")
|
|
def list_notes(book_id: int) -> Response:
|
|
_require_book(book_id)
|
|
rows = query(
|
|
f"SELECT {NOTE_COLUMNS} FROM notes n WHERE n.book_id = %s ORDER BY n.created_at DESC, n.id DESC",
|
|
(book_id,),
|
|
)
|
|
return jsonify([to_json(row) for row in rows])
|
|
|
|
|
|
@bp.post("/books/<int(max=2147483647):book_id>/notes")
|
|
def add_note(book_id: int) -> ResponseReturnValue:
|
|
_require_book(book_id)
|
|
body = text(json_body({"body"}), "body", MAX_BODY)
|
|
row = query_row(
|
|
f"INSERT INTO notes AS n (book_id, body) VALUES (%s, %s) RETURNING {NOTE_COLUMNS}",
|
|
(book_id, body),
|
|
)
|
|
return jsonify(to_json(row)), 201
|
|
|
|
|
|
@bp.patch("/notes/<int(max=2147483647):note_id>")
|
|
def update_note(note_id: int) -> Response:
|
|
body = text(json_body({"body"}), "body", MAX_BODY)
|
|
row = query_one(
|
|
f"""
|
|
UPDATE notes n
|
|
SET body = %s, updated_at = now()
|
|
FROM books b
|
|
WHERE n.id = %s AND b.id = n.book_id AND b.user_id = %s
|
|
RETURNING {NOTE_COLUMNS}
|
|
""",
|
|
(body, note_id, g.user_id),
|
|
)
|
|
if row is None:
|
|
raise ApiError(404, "Note not found")
|
|
return jsonify(to_json(row))
|
|
|
|
|
|
@bp.delete("/notes/<int(max=2147483647):note_id>")
|
|
def delete_note(note_id: int) -> ResponseReturnValue:
|
|
row = query_one(
|
|
"DELETE FROM notes n USING books b WHERE n.id = %s AND b.id = n.book_id AND b.user_id = %s RETURNING n.id",
|
|
(note_id, g.user_id),
|
|
)
|
|
if row is None:
|
|
raise ApiError(404, "Note not found")
|
|
return "", 204
|