diff --git a/helpers/book_helpers.py b/helpers/book_helpers.py new file mode 100644 index 0000000..e627378 --- /dev/null +++ b/helpers/book_helpers.py @@ -0,0 +1,92 @@ +from typing import List, Optional +from datetime import datetime +from pydantic import BaseModel +from sqlalchemy.orm import Session +from models.book import Book +from schemas.book import BookCreate, BookUpdate + +def get_books(db: Session, skip: int = 0, limit: int = 100) -> List[Book]: + """ + Retrieve a list of books from the database. + + Args: + db (Session): SQLAlchemy database session + skip (int, optional): Number of books to skip. Defaults to 0. + limit (int, optional): Maximum number of books to return. Defaults to 100. + + Returns: + List[Book]: List of Book objects + """ + return db.query(Book).offset(skip).limit(limit).all() + +def get_book_by_id(db: Session, book_id: int) -> Optional[Book]: + """ + Retrieve a book by its ID from the database. + + Args: + db (Session): SQLAlchemy database session + book_id (int): ID of the book to retrieve + + Returns: + Optional[Book]: Book object if found, None otherwise + """ + return db.query(Book).filter(Book.id == book_id).first() + +def create_book(db: Session, book: BookCreate) -> Book: + """ + Create a new book in the database. + + Args: + db (Session): SQLAlchemy database session + book (BookCreate): Book data to create + + Returns: + Book: Created Book object + """ + db_book = Book(**book.dict()) + db.add(db_book) + db.commit() + db.refresh(db_book) + return db_book + +def update_book(db: Session, book_id: int, book_data: BookUpdate) -> Optional[Book]: + """ + Update an existing book in the database. + + Args: + db (Session): SQLAlchemy database session + book_id (int): ID of the book to update + book_data (BookUpdate): Updated book data + + Returns: + Optional[Book]: Updated Book object if found, None otherwise + """ + db_book = db.query(Book).filter(Book.id == book_id).first() + if not db_book: + return None + + for field, value in book_data.dict(exclude_unset=True).items(): + setattr(db_book, field, value) + + db.commit() + db.refresh(db_book) + return db_book + +def delete_book(db: Session, book_id: int) -> bool: + """ + Delete a book from the database. + + Args: + db (Session): SQLAlchemy database session + book_id (int): ID of the book to delete + + Returns: + bool: True if the book was deleted, False otherwise + """ + db_book = db.query(Book).filter(Book.id == book_id).first() + if not db_book: + return False + + db.delete(db_book) + db.commit() + return True \ No newline at end of file