diff --git a/helpers/book_helpers.py b/helpers/book_helpers.py index f2b3987..cb616b2 100644 --- a/helpers/book_helpers.py +++ b/helpers/book_helpers.py @@ -1,139 +1,118 @@ -from typing import Optional, Dict, Union, List -from datetime import datetime +from typing import List, Dict, Optional, Union, Any from sqlalchemy.orm import Session -import re +from sqlalchemy import and_, or_ 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). + Validate ISBN format. Args: isbn: ISBN string to validate Returns: - bool: True if valid ISBN format, False otherwise + bool: True if ISBN format is valid, 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)) + # Remove any hyphens or spaces + isbn = isbn.replace('-', '').replace(' ', '') + + if len(isbn) == 13: # ISBN-13 + return isbn.isdigit() + elif len(isbn) == 10: # ISBN-10 + return isbn[:-1].isdigit() and (isbn[-1].isdigit() or isbn[-1].lower() == 'x') return False -def get_book_by_isbn(db: Session, isbn: str) -> Optional[Book]: +def get_available_books(db: Session, skip: int = 0, limit: int = 100) -> List[Book]: """ - Get a book by ISBN. + Get list of available books with pagination. Args: db: Database session - isbn: ISBN to search for + skip: Number of records to skip + limit: Maximum number of records to return Returns: - Book object if found, None otherwise + List of available Book objects """ - return db.query(Book).filter(Book.isbn == isbn).first() + return db.query(Book)\ + .filter(Book.is_available == True)\ + .filter(Book.quantity > 0)\ + .offset(skip)\ + .limit(limit)\ + .all() def search_books( - db: Session, - title: Optional[str] = None, - author: Optional[str] = None, - publisher: Optional[str] = None, - available_only: bool = False + db: Session, + search_term: str, + include_unavailable: bool = False ) -> List[Book]: """ - Search books with various filters. + Search books by title, author or ISBN. 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 + search_term: Search string + include_unavailable: Whether to include unavailable 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 + # Create search conditions + search_conditions = or_( + Book.title.ilike(f"%{search_term}%"), + Book.author.ilike(f"%{search_term}%"), + Book.isbn == search_term ) - db.add(db_book) - db.commit() - db.refresh(db_book) + if not include_unavailable: + query = query.filter(Book.is_available == True) - return db_book + return query.filter(search_conditions).all() -def update_book_availability(db: Session, book_id: int, copies_change: int) -> Union[Book, Dict[str, str]]: +def update_book_availability(db: Session, book_id: int) -> Optional[Book]: """ - Update book availability and copies count. + Update book availability based on quantity. 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 + Updated Book object if found, None otherwise """ 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 + if book: + book.is_available = book.quantity > 0 + db.commit() + db.refresh(book) + return book + return None + +def validate_book_data(book_data: Union[BookCreate, BookUpdate]) -> Dict[str, str]: + """ + Validate book data before creation/update. - db.commit() - db.refresh(book) - return book \ No newline at end of file + Args: + book_data: Book data to validate + + Returns: + Dict containing any validation errors + """ + errors = {} + + if hasattr(book_data, 'isbn') and book_data.isbn: + if not validate_isbn(book_data.isbn): + errors['isbn'] = "Invalid ISBN format" + + if hasattr(book_data, 'price') and book_data.price: + if book_data.price < 0: + errors['price'] = "Price cannot be negative" + + if hasattr(book_data, 'quantity') and book_data.quantity: + if book_data.quantity < 0: + errors['quantity'] = "Quantity cannot be negative" + + return errors \ No newline at end of file