from typing import List, Tuple import time import logging from sqlalchemy.exc import IntegrityError, SQLAlchemyError, DatabaseError as SqlAlchemyDatabaseError from utils.errors.database import DatabaseError, DuplicateEntryError, DatabaseConnectionError from models import BookCategoryStatisticsOverview from database.manager import DatabaseManager from database import get_total_count from utils.config import UserConfig logger = logging.getLogger(__name__) def fetch_all_book_category_statistics_overviews() -> List[BookCategoryStatisticsOverview]: with DatabaseManager.get_session() as session: try: return session.query(BookCategoryStatisticsOverview).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 occurred when fetching all category overviews: {e}") raise DatabaseError("An error occurred when fetching all category overviews") from e def fetch_all_book_category_statistics_overviews_with_count() -> Tuple[List[BookCategoryStatisticsOverview], int]: with DatabaseManager.get_session() as session: try: category_list = session.query(BookCategoryStatisticsOverview).all() user_config = UserConfig() if user_config.simulate_slowdown: logger.debug("Simulating slowdown after fetching statistics for 10 seconds") time.sleep(10) else: logger.debug("Performing category statistics update normally") count = get_total_count(session) return (category_list, count) 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 occurred when fetching all category overviews: {e}") raise DatabaseError("An error occurred when fetching all category overviews") from e __all__ = ["fetch_all_book_category_statistics_overviews", "fetch_all_book_category_statistics_overviews_with_count"]