Fix unique constraint on book table to set uniqueness over user_id and isbn

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)

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
 

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
 

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
 

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: