156 lines
4.5 KiB
Python
156 lines
4.5 KiB
Python
from typing import List, Dict, Optional, Union, Any
|
|
from datetime import datetime
|
|
import re
|
|
from sqlalchemy.orm import Session
|
|
from models.book import Book
|
|
from schemas.book import BookCreate, BookUpdate
|
|
|
|
def validate_isbn(isbn: str) -> bool:
|
|
"""
|
|
Validate ISBN format (both ISBN-10 and ISBN-13).
|
|
|
|
Args:
|
|
isbn: ISBN string to validate
|
|
|
|
Returns:
|
|
bool: True if ISBN format is valid, False otherwise
|
|
"""
|
|
# Remove hyphens and spaces
|
|
isbn = isbn.replace('-', '').replace(' ', '')
|
|
|
|
if len(isbn) == 10:
|
|
# ISBN-10 validation
|
|
if not isbn[:-1].isdigit():
|
|
return False
|
|
check = sum((10 - i) * int(x) for i, x in enumerate(isbn[:9]))
|
|
check = (11 - (check % 11)) % 11
|
|
return str(check) == isbn[-1] or (check == 10 and isbn[-1].upper() == 'X')
|
|
|
|
elif len(isbn) == 13:
|
|
# ISBN-13 validation
|
|
if not isbn.isdigit():
|
|
return False
|
|
check = sum((3 if i % 2 else 1) * int(x) for i, x in enumerate(isbn[:12]))
|
|
check = (10 - (check % 10)) % 10
|
|
return int(isbn[-1]) == check
|
|
|
|
return False
|
|
|
|
def get_book_by_isbn(db: Session, isbn: str) -> Optional[Book]:
|
|
"""
|
|
Retrieve a book by its 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,
|
|
year_from: Optional[int] = None,
|
|
year_to: Optional[int] = None,
|
|
available_only: bool = False
|
|
) -> List[Book]:
|
|
"""
|
|
Search books with multiple optional filters.
|
|
|
|
Args:
|
|
db: Database session
|
|
title: Optional title search string
|
|
author: Optional author search string
|
|
publisher: Optional publisher search string
|
|
year_from: Optional start year for publication
|
|
year_to: Optional end year for publication
|
|
available_only: If True, return only available books
|
|
|
|
Returns:
|
|
List of Book objects matching the search criteria
|
|
"""
|
|
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 year_from:
|
|
query = query.filter(Book.publication_year >= year_from)
|
|
if year_to:
|
|
query = query.filter(Book.publication_year <= year_to)
|
|
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": "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,
|
|
description=book_data.description,
|
|
publication_year=book_data.publication_year,
|
|
pages=book_data.pages,
|
|
publisher=book_data.publisher,
|
|
is_available=book_data.is_available
|
|
)
|
|
|
|
db.add(db_book)
|
|
db.commit()
|
|
db.refresh(db_book)
|
|
|
|
return db_book
|
|
|
|
def update_book_availability(db: Session, isbn: str, is_available: bool) -> Union[Book, Dict[str, str]]:
|
|
"""
|
|
Update the availability status of a book.
|
|
|
|
Args:
|
|
db: Database session
|
|
isbn: ISBN of the book to update
|
|
is_available: New availability status
|
|
|
|
Returns:
|
|
Updated Book object if successful, error dict otherwise
|
|
"""
|
|
book = get_book_by_isbn(db, isbn)
|
|
if not book:
|
|
return {"error": "Book not found"}
|
|
|
|
book.is_available = is_available
|
|
db.commit()
|
|
db.refresh(book)
|
|
|
|
return book |