feat: add endpoint for updating book records in database
This commit is contained in:
parent
8e8910498a
commit
456f3bb8a7
22
alembic/versions/20250429_184238_743ffe0b_update_book.py
Normal file
22
alembic/versions/20250429_184238_743ffe0b_update_book.py
Normal file
@ -0,0 +1,22 @@
|
||||
"""Add a new update_book helper function
|
||||
Revision ID: 4a8c3b7d9e2f
|
||||
Revises: 3c1332db
|
||||
Create Date: 2023-11-15 10:25:00.000000
|
||||
"""
|
||||
|
||||
# revision identifiers, used by Alembic.
|
||||
revision = '4a8c3b7d9e2f'
|
||||
down_revision = '3c1332db'
|
||||
branch_labels = None
|
||||
depends_on = None
|
||||
|
||||
|
||||
def upgrade():
|
||||
# This migration only adds a helper method to the Book model
|
||||
# No database schema changes are needed
|
||||
pass
|
||||
|
||||
|
||||
def downgrade():
|
||||
# No database schema changes to revert
|
||||
pass
|
@ -0,0 +1,47 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from sqlalchemy.orm import Session
|
||||
from uuid import UUID
|
||||
|
||||
from core.database import get_db
|
||||
from schemas.book import BookSchema, BookUpdate
|
||||
from helpers.book_helpers import update_book, get_book_by_id
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
@router.put("/books", response_model=BookSchema)
|
||||
async def update_book_endpoint(
|
||||
book_id: UUID,
|
||||
book_data: BookUpdate,
|
||||
db: Session = Depends(get_db)
|
||||
):
|
||||
"""
|
||||
Update an existing book in the database.
|
||||
|
||||
Args:
|
||||
book_id: UUID of the book to update
|
||||
book_data: Updated book data
|
||||
db: Database session
|
||||
|
||||
Returns:
|
||||
The updated book
|
||||
|
||||
Raises:
|
||||
HTTPException: If the book is not found
|
||||
"""
|
||||
# First check if the book exists
|
||||
book = get_book_by_id(db, book_id)
|
||||
if not book:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"Book with ID {book_id} not found"
|
||||
)
|
||||
|
||||
# Update the book
|
||||
updated_book = update_book(db, book_id, book_data)
|
||||
if not updated_book:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_400_BAD_REQUEST,
|
||||
detail="Failed to update book"
|
||||
)
|
||||
|
||||
return updated_book
|
@ -3,7 +3,7 @@ from sqlalchemy.orm import Session
|
||||
from uuid import UUID
|
||||
|
||||
from models.book import Book
|
||||
from schemas.book import BookCreate
|
||||
from schemas.book import BookCreate, BookUpdate
|
||||
|
||||
def create_book(db: Session, book_data: BookCreate) -> Book:
|
||||
"""
|
||||
@ -22,6 +22,30 @@ def create_book(db: Session, book_data: BookCreate) -> Book:
|
||||
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 get_book_by_id(db: Session, book_id: UUID) -> Optional[Book]:
|
||||
"""
|
||||
Retrieves a book by its ID.
|
||||
|
@ -3,6 +3,7 @@ from sqlalchemy.dialects.postgresql import UUID
|
||||
from sqlalchemy.sql import func
|
||||
from core.database import Base
|
||||
import uuid
|
||||
from typing import Optional
|
||||
|
||||
class Book(Base):
|
||||
__tablename__ = "books"
|
||||
@ -17,3 +18,29 @@ class Book(Base):
|
||||
# Timestamps
|
||||
created_at = Column(DateTime, default=func.now())
|
||||
updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
|
||||
|
||||
@classmethod
|
||||
def update_book(cls, db_session, book_id: uuid.UUID, book_update) -> Optional["Book"]:
|
||||
"""
|
||||
Update an existing book record in the database using the BookUpdate schema.
|
||||
|
||||
Args:
|
||||
db_session: The database session
|
||||
book_id: UUID of the book to update
|
||||
book_update: BookUpdate schema instance with fields to update
|
||||
|
||||
Returns:
|
||||
Updated Book instance or None if book not found
|
||||
"""
|
||||
book = db_session.query(cls).filter(cls.id == book_id).first()
|
||||
if not book:
|
||||
return None
|
||||
|
||||
# Update book attributes from the update schema
|
||||
update_data = book_update.dict(exclude_unset=True)
|
||||
for key, value in update_data.items():
|
||||
setattr(book, key, value)
|
||||
|
||||
db_session.commit()
|
||||
db_session.refresh(book)
|
||||
return book
|
Loading…
x
Reference in New Issue
Block a user