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 ]