Add helper functions for Book
This commit is contained in:
parent
e17e570d8a
commit
267099a1af
144
helpers/book_helpers.py
Normal file
144
helpers/book_helpers.py
Normal file
@ -0,0 +1,144 @@
|
|||||||
|
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
|
||||||
|
}
|
Loading…
x
Reference in New Issue
Block a user