Add helper functions for Book

This commit is contained in:
Backend IM Bot 2025-03-28 10:27:24 +00:00
parent 8433c8a1de
commit 12a895f04c

View File

@ -1,139 +1,118 @@
from typing import Optional, Dict, Union, List from typing import List, Dict, Optional, Union, Any
from datetime import datetime
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
import re from sqlalchemy import and_, or_
from models.book import Book from models.book import Book
from schemas.book import BookCreate, BookUpdate from schemas.book import BookCreate, BookUpdate
def validate_isbn(isbn: str) -> bool: def validate_isbn(isbn: str) -> bool:
""" """
Validate ISBN format (ISBN-10 or ISBN-13). Validate ISBN format.
Args: Args:
isbn: ISBN string to validate isbn: ISBN string to validate
Returns: Returns:
bool: True if valid ISBN format, False otherwise bool: True if ISBN format is valid, False otherwise
""" """
isbn = isbn.replace("-", "").replace(" ", "") # Remove any hyphens or spaces
if len(isbn) == 10: isbn = isbn.replace('-', '').replace(' ', '')
return bool(re.match(r'^\d{9}[\dX]$', isbn))
elif len(isbn) == 13: if len(isbn) == 13: # ISBN-13
return bool(re.match(r'^\d{13}$', isbn)) return isbn.isdigit()
elif len(isbn) == 10: # ISBN-10
return isbn[:-1].isdigit() and (isbn[-1].isdigit() or isbn[-1].lower() == 'x')
return False 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: Args:
db: Database session db: Database session
isbn: ISBN to search for skip: Number of records to skip
limit: Maximum number of records to return
Returns: 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( def search_books(
db: Session, db: Session,
title: Optional[str] = None, search_term: str,
author: Optional[str] = None, include_unavailable: bool = False
publisher: Optional[str] = None,
available_only: bool = False
) -> List[Book]: ) -> List[Book]:
""" """
Search books with various filters. Search books by title, author or ISBN.
Args: Args:
db: Database session db: Database session
title: Optional title to search search_term: Search string
author: Optional author to search include_unavailable: Whether to include unavailable books
publisher: Optional publisher to search
available_only: If True, return only available books
Returns: Returns:
List of matching Book objects List of matching Book objects
""" """
query = db.query(Book) query = db.query(Book)
if title: # Create search conditions
query = query.filter(Book.title.ilike(f"%{title}%")) search_conditions = or_(
if author: Book.title.ilike(f"%{search_term}%"),
query = query.filter(Book.author.ilike(f"%{author}%")) Book.author.ilike(f"%{search_term}%"),
if publisher: Book.isbn == search_term
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) if not include_unavailable:
db.commit() query = query.filter(Book.is_available == True)
db.refresh(db_book)
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: Args:
db: Database session db: Database session
book_id: ID of book to update book_id: ID of book to update
copies_change: Number of copies to add (positive) or remove (negative)
Returns: Returns:
Updated Book object or error dict Updated Book object if found, None otherwise
""" """
book = db.query(Book).filter(Book.id == book_id).first() book = db.query(Book).filter(Book.id == book_id).first()
if not book: if book:
return {"error": "Book not found"} book.is_available = book.quantity > 0
db.commit()
new_copies = book.copies_available + copies_change db.refresh(book)
if new_copies < 0: return book
return {"error": "Cannot have negative copies"} return None
book.copies_available = new_copies def validate_book_data(book_data: Union[BookCreate, BookUpdate]) -> Dict[str, str]:
book.is_available = new_copies > 0 """
Validate book data before creation/update.
db.commit() Args:
db.refresh(book) book_data: Book data to validate
return book
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