Add helper functions for Book

helpers/book_helpers.py

@@ -0,0 +1,139 @@
+from typing import Optional, Dict, Union, List
+from datetime import datetime
+from sqlalchemy.orm import Session
+import re
+from models.book import Book
+from schemas.book import BookCreate, BookUpdate
+
+def validate_isbn(isbn: str) -> bool:
+    """
+    Validate ISBN format (ISBN-10 or ISBN-13).
+    
+    Args:
+        isbn: ISBN string to validate
+        
+    Returns:
+        bool: True if valid ISBN format, False otherwise
+    """
+    isbn = isbn.replace("-", "").replace(" ", "")
+    if len(isbn) == 10:
+        return bool(re.match(r'^\d{9}[\dX]$', isbn))
+    elif len(isbn) == 13:
+        return bool(re.match(r'^\d{13}$', isbn))
+    return False
+
+def get_book_by_isbn(db: Session, isbn: str) -> Optional[Book]:
+    """
+    Get a book by ISBN.
+    
+    Args:
+        db: Database session
+        isbn: ISBN to search for
+        
+    Returns:
+        Book object if found, None otherwise
+    """
+    return db.query(Book).filter(Book.isbn == isbn).first()
+
+def search_books(
+    db: Session,
+    title: Optional[str] = None,
+    author: Optional[str] = None,
+    publisher: Optional[str] = None,
+    available_only: bool = False
+) -> List[Book]:
+    """
+    Search books with various filters.
+    
+    Args:
+        db: Database session
+        title: Optional title to search
+        author: Optional author to search
+        publisher: Optional publisher to search
+        available_only: If True, return only available books
+        
+    Returns:
+        List of matching Book objects
+    """
+    query = db.query(Book)
+    
+    if title:
+        query = query.filter(Book.title.ilike(f"%{title}%"))
+    if author:
+        query = query.filter(Book.author.ilike(f"%{author}%"))
+    if publisher:
+        query = query.filter(Book.publisher.ilike(f"%{publisher}%"))
+    if available_only:
+        query = query.filter(Book.is_available == True)
+        
+    return query.all()
+
+def create_book_safely(db: Session, book_data: BookCreate) -> Union[Book, Dict[str, str]]:
+    """
+    Create a new book with validation and error handling.
+    
+    Args:
+        db: Database session
+        book_data: Book data for creation
+        
+    Returns:
+        Book object if created successfully, error dict otherwise
+    """
+    # Validate ISBN
+    if not validate_isbn(book_data.isbn):
+        return {"error": "Invalid ISBN format"}
+        
+    # Check if book already exists
+    existing_book = get_book_by_isbn(db, book_data.isbn)
+    if existing_book:
+        return {"error": "Book with this ISBN already exists"}
+        
+    # Validate publication year
+    current_year = datetime.now().year
+    if book_data.publication_year > current_year:
+        return {"error": "Publication year cannot be in the future"}
+        
+    # Create the book
+    db_book = Book(
+        title=book_data.title,
+        author=book_data.author,
+        isbn=book_data.isbn,
+        publication_year=book_data.publication_year,
+        publisher=book_data.publisher,
+        description=book_data.description,
+        copies_available=book_data.copies_available,
+        is_available=book_data.copies_available > 0
+    )
+    
+    db.add(db_book)
+    db.commit()
+    db.refresh(db_book)
+    
+    return db_book
+
+def update_book_availability(db: Session, book_id: int, copies_change: int) -> Union[Book, Dict[str, str]]:
+    """
+    Update book availability and copies count.
+    
+    Args:
+        db: Database session
+        book_id: ID of book to update
+        copies_change: Number of copies to add (positive) or remove (negative)
+        
+    Returns:
+        Updated Book object or error dict
+    """
+    book = db.query(Book).filter(Book.id == book_id).first()
+    if not book:
+        return {"error": "Book not found"}
+        
+    new_copies = book.copies_available + copies_change
+    if new_copies < 0:
+        return {"error": "Cannot have negative copies"}
+        
+    book.copies_available = new_copies
+    book.is_available = new_copies > 0
+    
+    db.commit()
+    db.refresh(book)
+    return book