92 lines
2.5 KiB
Python
92 lines
2.5 KiB
Python
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 |