66 lines
2.1 KiB
Python
66 lines
2.1 KiB
Python
from sqlalchemy import Column, String, DateTime
|
|
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"
|
|
|
|
# Primary key using UUID
|
|
id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
|
|
|
|
# Book details
|
|
title = Column(String, nullable=False, index=True)
|
|
author = Column(String, nullable=False, index=True)
|
|
|
|
# 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
|
|
|
|
@classmethod
|
|
def delete_book(cls, db_session, book_id: uuid.UUID) -> bool:
|
|
"""
|
|
Delete a book record from the database.
|
|
|
|
Args:
|
|
db_session: The database session
|
|
book_id: UUID of the book to delete
|
|
|
|
Returns:
|
|
True if book was successfully deleted, False if book not found
|
|
"""
|
|
book = db_session.query(cls).filter(cls.id == book_id).first()
|
|
if not book:
|
|
return False
|
|
|
|
db_session.delete(book)
|
|
db_session.commit()
|
|
return True |