133 lines
4.3 KiB
Python
133 lines
4.3 KiB
Python
from typing import Optional, List
|
|
from sqlalchemy.orm import Session
|
|
from uuid import UUID
|
|
|
|
from models.book import Book
|
|
from schemas.book import BookCreate, BookUpdate
|
|
|
|
def create_book(db: Session, book_data: BookCreate) -> Book:
|
|
"""
|
|
Creates a new book in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_data (BookCreate): The data for the book to create.
|
|
|
|
Returns:
|
|
Book: The newly created book object.
|
|
"""
|
|
db_book = Book(**book_data.dict())
|
|
db.add(db_book)
|
|
db.commit()
|
|
db.refresh(db_book)
|
|
return db_book
|
|
|
|
def update_book(db: Session, book_id: UUID, book_data: BookUpdate) -> Optional[Book]:
|
|
"""
|
|
Updates an existing book in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_id (UUID): The ID of the book to update.
|
|
book_data (BookUpdate): The data to update the book with.
|
|
|
|
Returns:
|
|
Optional[Book]: The updated book object if found, otherwise None.
|
|
"""
|
|
book = db.query(Book).filter(Book.id == book_id).first()
|
|
if not book:
|
|
return None
|
|
|
|
update_data = book_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(book, key, value)
|
|
|
|
db.commit()
|
|
db.refresh(book)
|
|
return book
|
|
|
|
def delete_book(db: Session, book_id: UUID) -> bool:
|
|
"""
|
|
Deletes a book from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_id (UUID): The ID of the book to delete.
|
|
|
|
Returns:
|
|
bool: True if the book was successfully deleted, False if the book was not found.
|
|
"""
|
|
book = db.query(Book).filter(Book.id == book_id).first()
|
|
if not book:
|
|
return False
|
|
|
|
db.delete(book)
|
|
db.commit()
|
|
return True
|
|
|
|
def get_book_by_id(db: Session, book_id: UUID) -> Optional[Book]:
|
|
"""
|
|
Retrieves a book by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_id (UUID): The ID of the book to retrieve.
|
|
|
|
Returns:
|
|
Optional[Book]: The book object if found, otherwise None.
|
|
"""
|
|
return db.query(Book).filter(Book.id == book_id).first()
|
|
|
|
def get_books_by_author(db: Session, author: str, skip: int = 0, limit: int = 100) -> List[Book]:
|
|
"""
|
|
Retrieves books by a specific author.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
author (str): The author name to filter by.
|
|
skip (int, optional): Number of records to skip for pagination. Defaults to 0.
|
|
limit (int, optional): Maximum number of records to return. Defaults to 100.
|
|
|
|
Returns:
|
|
List[Book]: A list of books by the specified author.
|
|
"""
|
|
return db.query(Book).filter(Book.author == author).offset(skip).limit(limit).all()
|
|
|
|
def get_books_by_title(db: Session, title: str, skip: int = 0, limit: int = 100) -> List[Book]:
|
|
"""
|
|
Retrieves books by a specific title or partial title match.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
title (str): The title or partial title to search for.
|
|
skip (int, optional): Number of records to skip for pagination. Defaults to 0.
|
|
limit (int, optional): Maximum number of records to return. Defaults to 100.
|
|
|
|
Returns:
|
|
List[Book]: A list of books matching the title search.
|
|
"""
|
|
return db.query(Book).filter(Book.title.ilike(f"%{title}%")).offset(skip).limit(limit).all()
|
|
|
|
def get_all_books(db: Session, author: Optional[str] = None, title: Optional[str] = None, skip: int = 0, limit: int = 100) -> List[Book]:
|
|
"""
|
|
Retrieves all books with optional filtering by author or title and pagination.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
author (Optional[str], optional): The author name to filter by. Defaults to None.
|
|
title (Optional[str], optional): The title or partial title to search for. Defaults to None.
|
|
skip (int, optional): Number of records to skip for pagination. Defaults to 0.
|
|
limit (int, optional): Maximum number of records to return. Defaults to 100.
|
|
|
|
Returns:
|
|
List[Book]: A list of book objects matching the filters.
|
|
"""
|
|
query = db.query(Book)
|
|
|
|
if author:
|
|
query = query.filter(Book.author == author)
|
|
|
|
if title:
|
|
query = query.filter(Book.title.ilike(f"%{title}%"))
|
|
|
|
return query.offset(skip).limit(limit).all() |