From 173caa9a9ab9350b79e4451c0ed1c34b66e7e377 Mon Sep 17 00:00:00 2001 From: codebude Date: Mon, 8 Jun 2026 07:56:52 +0200 Subject: [PATCH] Fix unique constraint on book table to set uniqueness over user_id and isbn --- .../f7e9d1c3b5a2_make_isbn_unique_per_user.py | 69 ++++++++++++++++ backend/app/models.py | 6 +- backend/app/routers/books.py | 2 +- backend/app/routers/import_.py | 2 +- backend/tests/test_books.py | 78 +++++++++++++++++++ 5 files changed, 154 insertions(+), 3 deletions(-) create mode 100644 backend/alembic/versions/f7e9d1c3b5a2_make_isbn_unique_per_user.py diff --git a/backend/alembic/versions/f7e9d1c3b5a2_make_isbn_unique_per_user.py b/backend/alembic/versions/f7e9d1c3b5a2_make_isbn_unique_per_user.py new file mode 100644 index 0000000..bf853a6 --- /dev/null +++ b/backend/alembic/versions/f7e9d1c3b5a2_make_isbn_unique_per_user.py @@ -0,0 +1,69 @@ +"""make isbn unique per user instead of globally + +Revision ID: f7e9d1c3b5a2 +Revises: 86fa9b4f6d61 +Create Date: 2026-06-08 07:30:00.000000 + +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "f7e9d1c3b5a2" +down_revision: Union[str, Sequence[str], None] = "86fa9b4f6d61" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def _recreate_book_table(conn, *, global_unique: bool) -> None: + """Recreate the book table with either a global or per-user ISBN unique constraint.""" + old_cols = [] + for col in sa.inspect(conn).get_columns("book"): + old_cols.append(col) + + col_names = [c["name"] for c in old_cols] + pk_cols = [c["name"] for c in old_cols if c.get("primary_key")] + pk_name = pk_cols[0] if pk_cols else "id" + + col_defs = [] + for c in old_cols: + typ = str(c["type"]) + nullable = "NOT NULL" if not c.get("nullable", True) else "" + col_defs.append(f" {c['name']} {typ} {nullable}".strip()) + + extras = [f"PRIMARY KEY ({pk_name})"] + for fk in sa.inspect(conn).get_foreign_keys("book"): + extras.append( + f"FOREIGN KEY({fk['constrained_columns'][0]}) REFERENCES {fk['referred_table']}({fk['referred_columns'][0]})" + ) + extras.append("UNIQUE (isbn)" if global_unique else "UNIQUE (user_id, isbn)") + + col_list = ", ".join(col_names) + extra_sql = ",\n ".join(extras) + + sql = f"""CREATE TABLE book__new ( + {',\n '.join(col_defs)}, + {extra_sql} +)""" + conn.execute(sa.text(sql)) + conn.execute(sa.text(f"INSERT INTO book__new ({col_list}) SELECT {col_list} FROM book")) + + for idx in sa.inspect(conn).get_indexes("book"): + idx_name = idx["name"] + if idx_name: + cols = ", ".join(idx["column_names"]) + unique = "UNIQUE " if idx.get("unique") else "" + conn.execute(sa.text(f"CREATE {unique}INDEX IF NOT EXISTS {idx_name} ON book__new ({cols})")) + + conn.execute(sa.text("DROP TABLE book")) + conn.execute(sa.text("ALTER TABLE book__new RENAME TO book")) + + +def upgrade() -> None: + _recreate_book_table(op.get_bind(), global_unique=False) + + +def downgrade() -> None: + _recreate_book_table(op.get_bind(), global_unique=True) diff --git a/backend/app/models.py b/backend/app/models.py index c04369c..3ea2198 100644 --- a/backend/app/models.py +++ b/backend/app/models.py @@ -54,11 +54,15 @@ class UserRole(str, Enum): class Book(SQLModel, table=True): """A book in the user's library.""" + __table_args__ = ( + sa.UniqueConstraint("user_id", "isbn", name="uq_book_user_id_isbn"), + ) + id: Optional[int] = Field(default=None, primary_key=True) title: str = Field(index=True) subtitle: Optional[str] = None author: str = Field(default="", index=True) - isbn: Optional[str] = Field(default=None, unique=True) + isbn: Optional[str] = Field(default=None) cover_url: Optional[str] = None publisher: Optional[str] = None diff --git a/backend/app/routers/books.py b/backend/app/routers/books.py index 5feead5..05f96f3 100644 --- a/backend/app/routers/books.py +++ b/backend/app/routers/books.py @@ -124,7 +124,7 @@ def _normalize_language(language: str | None) -> str | None: def _raise_integrity_conflict(exc: IntegrityError) -> None: """Convert ISBN unique-constraint violations to HTTP 409.""" message = str(exc.orig).lower() if exc.orig else str(exc).lower() - if "book.isbn" in message and "unique" in message: + if ("book.isbn" in message or "uq_book_user_id_isbn" in message) and "unique" in message: raise HTTPException(status_code=409, detail="This ISBN is already used by another book.") from exc raise diff --git a/backend/app/routers/import_.py b/backend/app/routers/import_.py index fe3a750..3a9fc93 100644 --- a/backend/app/routers/import_.py +++ b/backend/app/routers/import_.py @@ -27,7 +27,7 @@ def _raise_integrity_conflict(exc: IntegrityError) -> None: """Convert ISBN unique-constraint violations to HTTP 409.""" message = str(exc.orig).lower() if exc.orig else str(exc).lower() - if "book.isbn" in message and "unique" in message: + if ("book.isbn" in message or "uq_book_user_id_isbn" in message) and "unique" in message: raise HTTPException(status_code=409, detail="This ISBN is already used by another book.") from exc raise diff --git a/backend/tests/test_books.py b/backend/tests/test_books.py index 776704a..06c1f47 100644 --- a/backend/tests/test_books.py +++ b/backend/tests/test_books.py @@ -991,6 +991,84 @@ def test_suggest_user_isolation(client: TestClient, create_user_with_key: Callab assert resp2.json()["suggestions"] == [] +# ── user data isolation ──────────────────────────────────────────────────────── + +def test_same_isbn_allowed_for_different_users(client: TestClient, create_user_with_key: Callable[..., tuple[User, str]]) -> None: + """ISBN uniqueness is per-user — two users can each have the same ISBN.""" + + shared_isbn = "9780441013593" + + # Create book with a shared ISBN for User 1 (default admin) + book1 = _create_book(client, title="Dune", author="Frank Herbert", page_count=412, isbn=shared_isbn) + + # Create second user and their book with the same ISBN + user2, key2 = create_user_with_key(email="other@example.com") + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + resp = c2.post("/api/books", json={ + "title": "Dune", "author": "Frank Herbert", "page_count": 412, "isbn": shared_isbn, + }) + assert resp.status_code == 201 + book2 = resp.json() + + # ── list ── + list1 = client.get("/api/books").json() + assert len(list1["books"]) == 1 + assert list1["books"][0]["id"] == book1["id"] + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + list2 = c2.get("/api/books").json() + assert len(list2["books"]) == 1 + assert list2["books"][0]["id"] == book2["id"] + + # ── get by id (own) ── + assert client.get(f"/api/books/{book1['id']}").status_code == 200 + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + assert c2.get(f"/api/books/{book2['id']}").status_code == 200 + + # ── get by id (other user) ── + assert client.get(f"/api/books/{book2['id']}").status_code == 404 + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + assert c2.get(f"/api/books/{book1['id']}").status_code == 404 + + # ── update (other user) ── + assert client.patch(f"/api/books/{book2['id']}", json={"title": "Hacked"}).status_code == 404 + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + assert c2.patch(f"/api/books/{book1['id']}", json={"title": "Hacked"}).status_code == 404 + + # ── delete (other user) ── + assert client.delete(f"/api/books/{book2['id']}").status_code == 404 + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + assert c2.delete(f"/api/books/{book1['id']}").status_code == 404 + + # ── stats isolation ── + stats1 = client.get("/api/books/stats").json() + assert stats1["total_books"] == 1 + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + stats2 = c2.get("/api/books/stats").json() + assert stats2["total_books"] == 1 + + # ── suggestions isolation ── + resp1 = client.get("/api/books/suggestions/authors?q=Frank") + assert resp1.json()["suggestions"] == ["Frank Herbert"] + + with TestClient(client.app) as c2: + c2.headers.update({"X-API-Key": key2}) + resp2 = c2.get("/api/books/suggestions/authors?q=Frank") + assert resp2.json()["suggestions"] == ["Frank Herbert"] + + # ── prevent removing date_finished for read books ────────────────────────────── def test_update_book_rejects_clearing_date_finished_for_read(client: TestClient) -> None: