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