diff --git a/migrations/versions/5584e4fd820e_add_unique_constraint_to_isbn_field.py b/migrations/versions/5584e4fd820e_add_unique_constraint_to_isbn_field.py new file mode 100644 index 0000000..e7ed915 --- /dev/null +++ b/migrations/versions/5584e4fd820e_add_unique_constraint_to_isbn_field.py @@ -0,0 +1,64 @@ +"""Add unique constraint to ISBN field + +Revision ID: 5584e4fd820e +Revises: 0c155d83c55b +Create Date: 2026-03-31 13:41:16.356631 + +""" +from alembic import op +import sqlalchemy as sa + + +# revision identifiers, used by Alembic. +revision = '5584e4fd820e' +down_revision = '0c155d83c55b' +branch_labels = None +depends_on = None + + +def upgrade(): + # ### commands auto generated by Alembic - please adjust! ### + + # First, convert empty string ISBNs to a temporary placeholder + # This is needed because we can't set NULL while column is NOT NULL + op.execute("UPDATE book SET isbn = '__EMPTY_ISBN__' WHERE isbn = ''") + + # Remove duplicate ISBNs, keeping only the book with the lowest ID + op.execute(""" + DELETE FROM book + WHERE isbn != '__EMPTY_ISBN__' + AND isbn != '' + AND id NOT IN ( + SELECT MIN(id) + FROM book + WHERE isbn != '__EMPTY_ISBN__' + AND isbn != '' + GROUP BY isbn + ) + """) + + with op.batch_alter_table('book', schema=None) as batch_op: + # Make the column nullable + batch_op.alter_column('isbn', + existing_type=sa.VARCHAR(length=20), + nullable=True) + + # Now convert the placeholder values to NULL + op.execute("UPDATE book SET isbn = NULL WHERE isbn = '__EMPTY_ISBN__'") + + with op.batch_alter_table('book', schema=None) as batch_op: + # Add the unique constraint + batch_op.create_unique_constraint('uq_book_isbn', ['isbn']) + + # ### end Alembic commands ### + + +def downgrade(): + # ### commands auto generated by Alembic - please adjust! ### + with op.batch_alter_table('book', schema=None) as batch_op: + batch_op.drop_constraint('uq_book_isbn', type_='unique') + batch_op.alter_column('isbn', + existing_type=sa.VARCHAR(length=20), + nullable=False) + + # ### end Alembic commands ### diff --git a/src/hxbooks/cli.py b/src/hxbooks/cli.py index 2cd8c20..2c456b7 100644 --- a/src/hxbooks/cli.py +++ b/src/hxbooks/cli.py @@ -16,6 +16,7 @@ from flask import Flask from . import library from .app import create_app from .db import db +from .library import DuplicateISBNError from .models import Author, Book, Genre, Reading, User, Wishlist logger = logging.getLogger(__name__) @@ -129,6 +130,9 @@ def add_book( cover_image_url=cover_url, ) click.echo(f"Added book: {book.title} (ID: {book.id})") + except DuplicateISBNError as e: + click.echo(f"Error: {e}", err=True) + sys.exit(1) except Exception as e: logger.error(f"Error adding book '{title}': {e}", exc_info=True) click.echo(f"Error adding book: {e}", err=True) @@ -504,6 +508,9 @@ def import_book( click.echo( f"Imported book: {book.title} by {', '.join(a.name.title() for a in book.authors)} (ID: {book.id})" ) + except DuplicateISBNError as e: + click.echo(f"Error: {e}", err=True) + sys.exit(1) except Exception as e: click.echo(f"Error importing book: {e}", err=True) sys.exit(1) diff --git a/src/hxbooks/db.py b/src/hxbooks/db.py index f72f9f8..b582587 100644 --- a/src/hxbooks/db.py +++ b/src/hxbooks/db.py @@ -6,7 +6,7 @@ from sqlalchemy.orm import DeclarativeBase class Base(DeclarativeBase): ... -db = SQLAlchemy(model_class=Base) +db = SQLAlchemy(model_class=Base, session_options={"autoflush": False}) def init_app(app: Flask) -> None: diff --git a/src/hxbooks/library.py b/src/hxbooks/library.py index 5389708..571d6d9 100644 --- a/src/hxbooks/library.py +++ b/src/hxbooks/library.py @@ -16,6 +16,7 @@ import requests from flask import current_app from PIL import Image from sqlalchemy import ColumnElement, Select, and_, func, or_, select +from sqlalchemy.exc import IntegrityError from sqlalchemy.orm import InstrumentedAttribute, joinedload from hxbooks.search import IsOperatorValue, QueryParser, SortDirection, ValueT @@ -29,6 +30,20 @@ from .search import ComparisonOperator, Field, FieldFilter logger = logging.getLogger(__name__) +class DuplicateISBNError(ValueError): + """Raised when attempting to create or update a book with a duplicate ISBN.""" + + def __init__( + self, isbn: str, existing_book_id: int, existing_book_title: str + ) -> None: + self.isbn = isbn + self.existing_book_id = existing_book_id + self.existing_book_title = existing_book_title + super().__init__( + f"ISBN '{isbn}' is already used by book '{existing_book_title}' (ID: {existing_book_id})" + ) + + def create_book( title: str, owner_id: int | None = None, @@ -52,7 +67,7 @@ def create_book( book = Book( title=title, owner_id=owner_id, - isbn=isbn or "", + isbn=isbn, publisher=publisher or "", edition=edition or "", description=description or "", @@ -79,7 +94,20 @@ def create_book( genre = _get_or_create_genre(genre_name) book.genres.append(genre) - db.session.commit() + try: + db.session.commit() + except IntegrityError as e: + db.session.rollback() + if "book.isbn" in str(e.orig).lower(): + # Find the existing book with this ISBN + existing_book = db.session.execute( + select(Book).filter(Book.isbn == isbn) + ).scalar_one_or_none() + if existing_book: + raise DuplicateISBNError( + isbn or "", existing_book.id, existing_book.title + ) from e + raise # Handle cover image if provided if cover_image_url: @@ -137,7 +165,6 @@ def update_book( assert title is not None, "Title is required when set_all_fields is True" book.title = title if isbn is not None or set_all_fields: - assert isbn is not None, "ISBN is required when set_all_fields is True" book.isbn = isbn if publisher is not None or set_all_fields: assert publisher is not None, ( @@ -208,7 +235,21 @@ def update_book( else: remove_book_cover(book) - db.session.commit() + try: + db.session.commit() + except IntegrityError as e: + db.session.rollback() + if "book.isbn" in str(e.orig).lower(): + # Find the existing book with this ISBN (excluding current book) + existing_book = db.session.execute( + select(Book).filter(Book.isbn == isbn, Book.id != book.id) + ).scalar_one_or_none() + if existing_book: + raise DuplicateISBNError( + isbn or "", existing_book.id, existing_book.title + ) from e + raise + return book diff --git a/src/hxbooks/main.py b/src/hxbooks/main.py index dc9f7a4..2ac9c16 100644 --- a/src/hxbooks/main.py +++ b/src/hxbooks/main.py @@ -9,7 +9,7 @@ import os import tempfile from datetime import date from pathlib import Path -from typing import Annotated, Any, Literal +from typing import Annotated, Any from flask import ( Blueprint, @@ -41,6 +41,7 @@ from hxbooks.search import IsOperatorValue, SortDirection from . import library from .db import db +from .library import DuplicateISBNError bp = Blueprint("main", __name__) @@ -51,7 +52,7 @@ logger = logging.getLogger(__name__) # Pydantic validation models StrOrNone = Annotated[str | None, BeforeValidator(lambda v: v.strip() or None)] StripStr = Annotated[str, StringConstraints(strip_whitespace=True)] -ISBNOrEmpty = Annotated[ISBN | Literal[""], BeforeValidator(lambda v: v.strip() or "")] +ISBNOrNone = Annotated[ISBN | None, BeforeValidator(lambda v: v.strip() or None)] TextareaList = Annotated[ list[str], BeforeValidator( @@ -70,7 +71,7 @@ UrlOrNone = Annotated[AnyHttpUrl | None, BeforeValidator(lambda v: v.strip() or class BookFormData(BaseModel): title: StripStr = Field(min_length=1) owner: StrOrNone = None - isbn: ISBNOrEmpty = "" + isbn: ISBNOrNone = None authors: TextareaList = Field(default_factory=list) genres: TextareaList = Field(default_factory=list) first_published: IntOrNone = Field(default=None, le=2030) @@ -283,7 +284,7 @@ def create_book() -> ResponseReturnValue: owner_id=owner_id, authors=form_data.authors, genres=form_data.genres, - isbn=str(form_data.isbn), + isbn=form_data.isbn, publisher=form_data.publisher, edition=form_data.edition, description=form_data.description, @@ -300,6 +301,10 @@ def create_book() -> ResponseReturnValue: flash(f"Book '{form_data.title}' created successfully!", "success") return redirect(url_for("main.book_detail", book_id=book.id)) + except DuplicateISBNError as e: + flash(f"Error: {e}", "error") + return render_template("book/create.html.j2", form_data=request.form) + except ValidationError as e: _flash_validation_errors(e) @@ -357,6 +362,9 @@ def update_book(book_id: int) -> ResponseReturnValue: flash("Book updated successfully!", "success") + except DuplicateISBNError as e: + flash(f"Error: {e}", "error") + except ValidationError as e: # Format validation errors for display _flash_validation_errors(e) diff --git a/src/hxbooks/models.py b/src/hxbooks/models.py index 940e870..1a27f0f 100644 --- a/src/hxbooks/models.py +++ b/src/hxbooks/models.py @@ -56,7 +56,9 @@ class Book(db.Model): # ty:ignore[unsupported-base] first_published: Mapped[int | None] = mapped_column(default=None) edition: Mapped[str] = mapped_column(String(200), default="") publisher: Mapped[str] = mapped_column(String(200), default="") - isbn: Mapped[str] = mapped_column(String(20), default="") + isbn: Mapped[str | None] = mapped_column( + String(20), unique=True, nullable=True, default=None + ) notes: Mapped[str] = mapped_column(default="") added_date: Mapped[datetime] = mapped_column(default=datetime.now) bought_date: Mapped[date | None] = mapped_column(default=None) diff --git a/src/hxbooks/templates/components/book_form.html.j2 b/src/hxbooks/templates/components/book_form.html.j2 index 6c827fb..a4fb555 100644 --- a/src/hxbooks/templates/components/book_form.html.j2 +++ b/src/hxbooks/templates/components/book_form.html.j2 @@ -12,7 +12,7 @@
- +
@@ -73,7 +73,7 @@
+ placeholder="https://example.com/cover.jpg">