diff --git a/helpers/generic_helpers.py b/helpers/generic_helpers.py new file mode 100644 index 0000000..02fe9c4 --- /dev/null +++ b/helpers/generic_helpers.py @@ -0,0 +1,169 @@ +from typing import Optional, Dict, List, Union, Any +from datetime import datetime +from sqlalchemy.orm import Session +from sqlalchemy import or_ +from pydantic import BaseModel, validator + +class GenericValidator: + """Validator class for Generic entity""" + + @staticmethod + def validate_name(name: str) -> bool: + """ + Validate generic name format and length + + Args: + name: Name to validate + + Returns: + bool: True if valid, False otherwise + """ + if not name or len(name) < 2 or len(name) > 100: + return False + return bool(name.strip()) + + @staticmethod + def validate_status(status: str) -> bool: + """ + Validate generic status + + Args: + status: Status to validate + + Returns: + bool: True if valid, False otherwise + """ + valid_statuses = ['active', 'inactive', 'pending', 'archived'] + return status.lower() in valid_statuses + + +def format_generic_response(generic: Any) -> Dict: + """ + Format generic entity for API response + + Args: + generic: Generic entity to format + + Returns: + Dict containing formatted generic data + """ + return { + 'id': generic.id, + 'name': generic.name, + 'description': generic.description, + 'status': generic.status, + 'created_at': generic.created_at.isoformat() if generic.created_at else None, + 'updated_at': generic.updated_at.isoformat() if generic.updated_at else None + } + + +def search_generics( + db: Session, + search_term: str, + status: Optional[str] = None, + limit: int = 10, + offset: int = 0 +) -> List[Any]: + """ + Search generic entities by name or description + + Args: + db: Database session + search_term: Term to search for + status: Optional status filter + limit: Max number of results + offset: Number of results to skip + + Returns: + List of matching generic entities + """ + query = db.query(Generic) + + if search_term: + query = query.filter( + or_( + Generic.name.ilike(f"%{search_term}%"), + Generic.description.ilike(f"%{search_term}%") + ) + ) + + if status: + query = query.filter(Generic.status == status) + + return query.offset(offset).limit(limit).all() + + +def bulk_update_status( + db: Session, + generic_ids: List[int], + new_status: str +) -> Dict[str, Any]: + """ + Bulk update status for multiple generic entities + + Args: + db: Database session + generic_ids: List of generic IDs to update + new_status: New status to set + + Returns: + Dict with update results + """ + if not GenericValidator.validate_status(new_status): + return {'error': 'Invalid status value'} + + try: + updated = db.query(Generic).filter( + Generic.id.in_(generic_ids) + ).update( + {Generic.status: new_status}, + synchronize_session=False + ) + db.commit() + return { + 'success': True, + 'updated_count': updated + } + except Exception as e: + db.rollback() + return { + 'success': False, + 'error': str(e) + } + + +def archive_old_generics( + db: Session, + days_threshold: int = 90 +) -> Dict[str, Any]: + """ + Archive generic entities older than threshold + + Args: + db: Database session + days_threshold: Days after which to archive + + Returns: + Dict with archive results + """ + cutoff_date = datetime.utcnow() - timedelta(days=days_threshold) + + try: + archived = db.query(Generic).filter( + Generic.created_at < cutoff_date, + Generic.status != 'archived' + ).update( + {Generic.status: 'archived'}, + synchronize_session=False + ) + db.commit() + return { + 'success': True, + 'archived_count': archived + } + except Exception as e: + db.rollback() + return { + 'success': False, + 'error': str(e) + } \ No newline at end of file