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) }