97 lines
2.9 KiB
Python
97 lines
2.9 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
from sqlalchemy import or_
|
|
from models.book import Book
|
|
from schemas.book import BookSchema, BookCreate, BookUpdate
|
|
import uuid
|
|
|
|
def get_books(db: Session, search_term: Optional[str] = None) -> List[BookSchema]:
|
|
"""
|
|
Retrieves a list of books from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
search_term (Optional[str]): An optional search term to filter books by title or author.
|
|
|
|
Returns:
|
|
List[BookSchema]: A list of book objects matching the search criteria.
|
|
"""
|
|
query = db.query(Book)
|
|
if search_term:
|
|
query = query.filter(or_(
|
|
Book.title.ilike(f"%{search_term}%"),
|
|
Book.author.ilike(f"%{search_term}%")
|
|
))
|
|
books = query.all()
|
|
return [BookSchema.from_orm(book) for book in books]
|
|
|
|
def get_book_by_id(db: Session, book_id: uuid.UUID) -> Optional[BookSchema]:
|
|
"""
|
|
Retrieves a single book by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_id (uuid.UUID): The ID of the book to retrieve.
|
|
|
|
Returns:
|
|
Optional[BookSchema]: The book object if found, otherwise None.
|
|
"""
|
|
book = db.query(Book).filter(Book.id == book_id).first()
|
|
return BookSchema.from_orm(book) if book else None
|
|
|
|
def create_book(db: Session, book_data: BookCreate) -> BookSchema:
|
|
"""
|
|
Creates a new book in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_data (BookCreate): The data for the book to create.
|
|
|
|
Returns:
|
|
BookSchema: The newly created book object.
|
|
"""
|
|
db_book = Book(**book_data.dict())
|
|
db.add(db_book)
|
|
db.commit()
|
|
db.refresh(db_book)
|
|
return BookSchema.from_orm(db_book)
|
|
|
|
def update_book(db: Session, book_id: uuid.UUID, book_data: BookUpdate) -> Optional[BookSchema]:
|
|
"""
|
|
Updates an existing book in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_id (uuid.UUID): The ID of the book to update.
|
|
book_data (BookUpdate): The updated data for the book.
|
|
|
|
Returns:
|
|
Optional[BookSchema]: The updated book object if found, otherwise None.
|
|
"""
|
|
book = db.query(Book).filter(Book.id == book_id).first()
|
|
if book:
|
|
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 BookSchema.from_orm(book)
|
|
return None
|
|
|
|
def delete_book(db: Session, book_id: uuid.UUID) -> bool:
|
|
"""
|
|
Deletes a book from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
book_id (uuid.UUID): The ID of the book to delete.
|
|
|
|
Returns:
|
|
bool: True if the book was successfully deleted, False otherwise.
|
|
"""
|
|
book = db.query(Book).filter(Book.id == book_id).first()
|
|
if book:
|
|
db.delete(book)
|
|
db.commit()
|
|
return True
|
|
return False |