mo/mitlist
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add database error messages and improve exception handling in CRUD operations

Browse Source
This commit is contained in:
Mohamad.Elsena 2025-05-08 16:00:12 +02:00
parent 5186892df6
commit c7fdb60130
3 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
be/app/config.py
View File

@ -98,6 +98,12 @@ Organic Bananas
HTTP_500_DETAIL: str = "Internal Server Error" HTTP_500_DETAIL: str = "Internal Server Error"
HTTP_503_DETAIL: str = "Service Unavailable" HTTP_503_DETAIL: str = "Service Unavailable"
# --- Database Error Messages ---
DB_CONNECTION_ERROR: str = "Database connection error"
DB_INTEGRITY_ERROR: str = "Database integrity error"
DB_TRANSACTION_ERROR: str = "Database transaction error"
DB_QUERY_ERROR: str = "Database query error"
class Config: class Config:
env_file = ".env" env_file = ".env"
env_file_encoding = 'utf-8' env_file_encoding = 'utf-8'

4
be/app/core/exceptions.py
View File

@ -122,10 +122,10 @@ class DatabaseTransactionError(HTTPException):
class DatabaseQueryError(HTTPException): class DatabaseQueryError(HTTPException):
"""Raised when a database query fails.""" """Raised when a database query fails."""
def __init__(self): def __init__(self, detail: str = settings.DB_QUERY_ERROR):
super().__init__( super().__init__(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR, status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
detail=settings.DB_QUERY_ERROR detail=detail
) )
class OCRServiceUnavailableError(HTTPException): class OCRServiceUnavailableError(HTTPException):

16
be/app/crud/list.py
View File

@ -51,18 +51,22 @@ async def get_lists_for_user(db: AsyncSession, user_id: int) -> PyList[ListModel
) )
user_group_ids = group_ids_result.scalars().all() user_group_ids = group_ids_result.scalars().all()
query = select(ListModel).where( # Build conditions for the OR clause dynamically
or_( conditions = [
and_(ListModel.created_by_id == user_id, ListModel.group_id == None), and_(ListModel.created_by_id == user_id, ListModel.group_id.is_(None))
ListModel.group_id.in_(user_group_ids) ]
) if user_group_ids: # Only add the IN clause if there are group IDs
).order_by(ListModel.updated_at.desc()) conditions.append(ListModel.group_id.in_(user_group_ids))
query = select(ListModel).where(or_(*conditions)).order_by(ListModel.updated_at.desc())
result = await db.execute(query) result = await db.execute(query)
return result.scalars().all() return result.scalars().all()
except OperationalError as e: except OperationalError as e:
raise DatabaseConnectionError(f"Failed to connect to database: {str(e)}") raise DatabaseConnectionError(f"Failed to connect to database: {str(e)}")
except SQLAlchemyError as e: except SQLAlchemyError as e:
# It would be helpful to log the original error e here for more detailed debugging
# For example: logger.error(f"SQLAlchemyError in get_lists_for_user: {type(e).__name__} - {str(e)}")
raise DatabaseQueryError(f"Failed to query user lists: {str(e)}") raise DatabaseQueryError(f"Failed to query user lists: {str(e)}")
async def get_list_by_id(db: AsyncSession, list_id: int, load_items: bool = False) -> Optional[ListModel]: async def get_list_by_id(db: AsyncSession, list_id: int, load_items: bool = False) -> Optional[ListModel]: