Add helper functions for Exception

helpers/exception_helpers.py

@@ -0,0 +1,126 @@
+from typing import List, Dict, Optional, Union, Any
+from sqlalchemy.orm import Session
+from sqlalchemy import text
+from models.user import User
+from fastapi import HTTPException
+import logging
+from datetime import datetime
+
+def get_all_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]:
+    """
+    Retrieve a list of users with pagination.
+    
+    Args:
+        db: Database session
+        skip: Number of records to skip
+        limit: Maximum number of records to return
+        
+    Returns:
+        List of User objects
+    """
+    try:
+        return db.query(User).offset(skip).limit(limit).all()
+    except Exception as e:
+        logging.error(f"Error fetching users: {str(e)}")
+        raise HTTPException(status_code=500, detail="Error retrieving users")
+
+def filter_users_by_criteria(
+    db: Session,
+    search_term: Optional[str] = None,
+    active_only: bool = True,
+    created_after: Optional[datetime] = None
+) -> List[User]:
+    """
+    Filter users based on various criteria.
+    
+    Args:
+        db: Database session
+        search_term: Search by username or email
+        active_only: Filter only active users
+        created_after: Filter users created after date
+        
+    Returns:
+        Filtered list of User objects
+    """
+    query = db.query(User)
+    
+    if search_term:
+        query = query.filter(
+            (User.username.ilike(f"%{search_term}%")) |
+            (User.email.ilike(f"%{search_term}%"))
+        )
+    
+    if active_only:
+        query = query.filter(User.is_active == True)
+        
+    if created_after:
+        query = query.filter(User.created_at >= created_after)
+        
+    return query.all()
+
+def get_users_with_stats(db: Session) -> Dict[str, Any]:
+    """
+    Get users list with additional statistics.
+    
+    Args:
+        db: Database session
+        
+    Returns:
+        Dictionary containing users and stats
+    """
+    try:
+        users = db.query(User).all()
+        total_users = len(users)
+        active_users = len([u for u in users if u.is_active])
+        
+        return {
+            "users": users,
+            "total_count": total_users,
+            "active_count": active_users,
+            "inactive_count": total_users - active_users
+        }
+    except Exception as e:
+        logging.error(f"Error getting user stats: {str(e)}")
+        raise HTTPException(status_code=500, detail="Error retrieving user statistics")
+
+def validate_user_list_params(skip: int, limit: int) -> None:
+    """
+    Validate pagination parameters for user list.
+    
+    Args:
+        skip: Number of records to skip
+        limit: Maximum number of records to return
+        
+    Raises:
+        HTTPException if parameters are invalid
+    """
+    if skip < 0:
+        raise HTTPException(status_code=400, detail="Skip parameter cannot be negative")
+    
+    if limit < 1:
+        raise HTTPException(status_code=400, detail="Limit must be greater than 0")
+        
+    if limit > 100:
+        raise HTTPException(status_code=400, detail="Limit cannot exceed 100")
+
+def format_user_list_response(users: List[User]) -> List[Dict[str, Any]]:
+    """
+    Format user list for API response.
+    
+    Args:
+        users: List of User objects
+        
+    Returns:
+        List of formatted user dictionaries
+    """
+    return [
+        {
+            "id": user.id,
+            "username": user.username,
+            "email": user.email,
+            "is_active": user.is_active,
+            "created_at": user.created_at.isoformat(),
+            "last_login": user.last_login.isoformat() if user.last_login else None
+        }
+        for user in users
+    ]