144 lines
3.8 KiB
Python
144 lines
3.8 KiB
Python
from typing import List, Dict, Optional, Union, Any
|
|
from datetime import datetime
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import desc, asc
|
|
from models.book import Book
|
|
from schemas.book import BookCreate, BookResponse
|
|
|
|
def get_books_with_filters(
|
|
db: Session,
|
|
skip: int = 0,
|
|
limit: int = 100,
|
|
sort_by: str = "title",
|
|
sort_order: str = "asc",
|
|
title_search: Optional[str] = None,
|
|
min_price: Optional[float] = None,
|
|
max_price: Optional[float] = None
|
|
) -> List[Book]:
|
|
"""
|
|
Get books with various filter options.
|
|
|
|
Args:
|
|
db: Database session
|
|
skip: Number of records to skip
|
|
limit: Maximum number of records to return
|
|
sort_by: Field to sort by
|
|
sort_order: Sort direction (asc/desc)
|
|
title_search: Search string for title
|
|
min_price: Minimum price filter
|
|
max_price: Maximum price filter
|
|
|
|
Returns:
|
|
List of Book objects matching the criteria
|
|
"""
|
|
query = db.query(Book)
|
|
|
|
if title_search:
|
|
query = query.filter(Book.title.ilike(f"%{title_search}%"))
|
|
|
|
if min_price is not None:
|
|
query = query.filter(Book.price >= min_price)
|
|
|
|
if max_price is not None:
|
|
query = query.filter(Book.price <= max_price)
|
|
|
|
if sort_order == "desc":
|
|
query = query.order_by(desc(getattr(Book, sort_by)))
|
|
else:
|
|
query = query.order_by(asc(getattr(Book, sort_by)))
|
|
|
|
return query.offset(skip).limit(limit).all()
|
|
|
|
def get_book_by_id(db: Session, book_id: int) -> Optional[Book]:
|
|
"""
|
|
Get a single book by ID.
|
|
|
|
Args:
|
|
db: Database session
|
|
book_id: ID of the book to retrieve
|
|
|
|
Returns:
|
|
Book object if found, None otherwise
|
|
"""
|
|
return db.query(Book).filter(Book.id == book_id).first()
|
|
|
|
def format_book_response(book: Book) -> BookResponse:
|
|
"""
|
|
Format a book object into API response format.
|
|
|
|
Args:
|
|
book: Book object to format
|
|
|
|
Returns:
|
|
Formatted book response
|
|
"""
|
|
return BookResponse(
|
|
id=book.id,
|
|
title=book.title,
|
|
author=book.author,
|
|
price=book.price,
|
|
isbn=book.isbn,
|
|
created_at=book.created_at,
|
|
updated_at=book.updated_at
|
|
)
|
|
|
|
def validate_isbn(isbn: str) -> bool:
|
|
"""
|
|
Validate ISBN format.
|
|
|
|
Args:
|
|
isbn: ISBN string to validate
|
|
|
|
Returns:
|
|
bool: True if valid ISBN format, False otherwise
|
|
"""
|
|
# Remove hyphens and spaces
|
|
isbn = isbn.replace('-', '').replace(' ', '')
|
|
|
|
if len(isbn) == 13: # ISBN-13
|
|
try:
|
|
total = sum(
|
|
(3 if i % 2 else 1) * int(d)
|
|
for i, d in enumerate(isbn[:-1])
|
|
)
|
|
check_digit = (10 - (total % 10)) % 10
|
|
return check_digit == int(isbn[-1])
|
|
except ValueError:
|
|
return False
|
|
return False
|
|
|
|
def check_book_availability(
|
|
db: Session,
|
|
book_id: int,
|
|
quantity_requested: int = 1
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Check if a book is available in requested quantity.
|
|
|
|
Args:
|
|
db: Database session
|
|
book_id: ID of book to check
|
|
quantity_requested: Quantity being requested
|
|
|
|
Returns:
|
|
Dict containing availability status and message
|
|
"""
|
|
book = get_book_by_id(db, book_id)
|
|
if not book:
|
|
return {
|
|
"available": False,
|
|
"message": "Book not found"
|
|
}
|
|
|
|
if book.stock_quantity >= quantity_requested:
|
|
return {
|
|
"available": True,
|
|
"message": "Book available",
|
|
"stock_quantity": book.stock_quantity
|
|
}
|
|
|
|
return {
|
|
"available": False,
|
|
"message": f"Only {book.stock_quantity} copies available",
|
|
"stock_quantity": book.stock_quantity
|
|
} |