67 lines
2.3 KiB
Python
67 lines
2.3 KiB
Python
from typing import Dict, List
|
|
import logging
|
|
|
|
from sqlalchemy.exc import IntegrityError, SQLAlchemyError, DatabaseError as SqlAlchemyDatabaseError
|
|
|
|
from utils.errors.database import DatabaseError, DuplicateEntryError, DatabaseConnectionError
|
|
from models import Book
|
|
from database.manager import DatabaseManager
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def fetch_all():
|
|
with DatabaseManager.get_session() as session:
|
|
try:
|
|
return session.query(Book).all()
|
|
except SqlAlchemyDatabaseError as e:
|
|
logger.critical("Connection with database interrupted")
|
|
raise DatabaseConnectionError(
|
|
"Connection with database interrupted") from e
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"An error occured when fetching all books: {e}")
|
|
raise DatabaseError(
|
|
"An error occured when fetching all books") from e
|
|
|
|
|
|
def create_new_book(book: Book):
|
|
pass
|
|
|
|
|
|
def update_book(book: Book):
|
|
session = DatabaseManager.get_session()
|
|
try:
|
|
with session:
|
|
logger.debug(f"Updating book {book.title}")
|
|
existing_book = session.query(Book).get(book.id)
|
|
|
|
if not existing_book:
|
|
logger.warning(f"Book with id {book.id} not found")
|
|
raise DatabaseError("Book not found in the database")
|
|
|
|
existing_book.title = book.title
|
|
existing_book.description = book.description
|
|
existing_book.year_published = book.year_published
|
|
existing_book.isbn = book.isbn
|
|
|
|
session.commit()
|
|
logger.info(f"{book.title} successfully updated")
|
|
except IntegrityError as e:
|
|
logger.warning("Data already exists")
|
|
session.rollback()
|
|
raise DuplicateEntryError(
|
|
"Data already exists in the database") from e
|
|
except SqlAlchemyDatabaseError as e:
|
|
session.rollback()
|
|
logger.critical("Connection with database interrupted")
|
|
raise DatabaseConnectionError(
|
|
"Connection with database interrupted") from e
|
|
except SQLAlchemyError as e:
|
|
logger.error(f"An error occured when saving book: {e}")
|
|
session.rollback()
|
|
raise DatabaseError(
|
|
"An error occured when updating the book") from e
|
|
|
|
|
|
__all__ = ["create_new_book", "update_book"]
|