diff --git a/helpers/exception_helpers.py b/helpers/exception_helpers.py new file mode 100644 index 0000000..f693041 --- /dev/null +++ b/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 + ] \ No newline at end of file