Make ISBN field nullable and add unique constraint
This commit is contained in:
@@ -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 ###
|
||||
@@ -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)
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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)
|
||||
|
||||
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)
|
||||
|
||||
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
|
||||
|
||||
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -12,7 +12,7 @@
|
||||
</div>
|
||||
<div class="col-md-3">
|
||||
<label for="isbn" class="form-label">ISBN</label>
|
||||
<input type="text" class="form-control" id="isbn" name="isbn" value="{{ book.isbn if book else '' }}">
|
||||
<input type="text" class="form-control" id="isbn" name="isbn" value="{{ book.isbn if book and book.isbn else '' }}">
|
||||
</div>
|
||||
</div>
|
||||
|
||||
@@ -73,7 +73,7 @@
|
||||
<div class="col-md-6">
|
||||
<label for="cover_image_url" class="form-label">Replace with URL</label>
|
||||
<input type="url" class="form-control" id="cover_image_url" name="cover_image_url" value=""
|
||||
placeholder="https://example.com/cover.jpg or file:///path/to/image.jpg">
|
||||
placeholder="https://example.com/cover.jpg">
|
||||
</div>
|
||||
<div class="col-md-6">
|
||||
<label for="cover_image_file" class="form-label">Replace with Upload</label>
|
||||
|
||||
Reference in New Issue
Block a user