""" Management command to warm permission cache for critical users """ from django.core.management.base import BaseCommand, CommandError from django.utils import timezone from users.permission_cache_service import permission_cache from users.models import CustomUser import logging logger = logging.getLogger(__name__) class Command(BaseCommand): help = 'Warm permission cache for critical users to improve performance' def add_arguments(self, parser): parser.add_argument( '--users', type=int, default=50, help='Number of users to warm cache for (default: 50)' ) parser.add_argument( '--user-ids', nargs='+', help='Specific user IDs to warm cache for' ) parser.add_argument( '--roles', nargs='+', choices=['admin', 'team_leader', 'loan_officer', 'secretary', 'auditor'], help='Specific roles to warm cache for' ) parser.add_argument( '--pages', nargs='+', default=['loans', 'clients', 'reports', 'dashboard', 'repayments'], help='Pages to warm cache for (default: all common pages)' ) parser.add_argument( '--force', action='store_true', help='Force cache warming even if cache already exists' ) def handle(self, *args, **options): start_time = timezone.now() self.stdout.write( self.style.SUCCESS(f'Starting permission cache warming at {start_time}') ) try: if options['user_ids']: # Warm cache for specific users results = self._warm_specific_users( options['user_ids'], options['pages'], options['force'] ) elif options['roles']: # Warm cache for users with specific roles results = self._warm_users_by_role( options['roles'], options['pages'], options['users'], options['force'] ) else: # Warm cache for critical users results = self._warm_critical_users( options['users'], options['pages'], options['force'] ) # Display results self._display_results(results, start_time) except Exception as e: raise CommandError(f'Error warming permission cache: {e}') def _warm_specific_users(self, user_ids, pages, force): """Warm cache for specific user IDs""" results = { 'total_users': 0, 'successful_users': 0, 'failed_users': 0, 'total_pages_warmed': 0, 'total_permissions_cached': 0, 'total_actions_cached': 0, 'errors': [] } for user_id in user_ids: try: # Check if user exists user = CustomUser.objects.get(id=user_id) if not force: # Check if cache already exists existing_cache = permission_cache.get_user_page_permissions(user_id, pages[0]) if existing_cache: self.stdout.write( self.style.WARNING(f'Cache already exists for user {user_id}, skipping (use --force to override)') ) continue # Warm cache for this user user_results = permission_cache.warm_user_cache(user_id, pages) results['total_users'] += 1 if not user_results.get('errors'): results['successful_users'] += 1 self.stdout.write( self.style.SUCCESS(f'✓ Warmed cache for user {user_id} ({user.get_full_name()})') ) else: results['failed_users'] += 1 self.stdout.write( self.style.ERROR(f'✗ Failed to warm cache for user {user_id}: {user_results["errors"]}') ) results['errors'].extend(user_results['errors']) results['total_pages_warmed'] += user_results['pages_warmed'] results['total_permissions_cached'] += user_results['permissions_cached'] results['total_actions_cached'] += user_results['actions_cached'] except CustomUser.DoesNotExist: results['failed_users'] += 1 error_msg = f'User {user_id} not found' results['errors'].append(error_msg) self.stdout.write(self.style.ERROR(f'✗ {error_msg}')) except Exception as e: results['failed_users'] += 1 error_msg = f'Error warming cache for user {user_id}: {str(e)}' results['errors'].append(error_msg) self.stdout.write(self.style.ERROR(f'✗ {error_msg}')) return results def _warm_users_by_role(self, roles, pages, limit, force): """Warm cache for users with specific roles""" results = { 'total_users': 0, 'successful_users': 0, 'failed_users': 0, 'total_pages_warmed': 0, 'total_permissions_cached': 0, 'total_actions_cached': 0, 'errors': [] } users = CustomUser.objects.filter( role__in=roles, is_active=True ).order_by('-last_login')[:limit] self.stdout.write(f'Found {users.count()} users with roles {roles}') for user in users: try: if not force: # Check if cache already exists existing_cache = permission_cache.get_user_page_permissions(str(user.id), pages[0]) if existing_cache: self.stdout.write( self.style.WARNING(f'Cache already exists for user {user.id}, skipping') ) continue # Warm cache for this user user_results = permission_cache.warm_user_cache(str(user.id), pages) results['total_users'] += 1 if not user_results.get('errors'): results['successful_users'] += 1 self.stdout.write( self.style.SUCCESS(f'✓ Warmed cache for {user.get_full_name()} ({user.role})') ) else: results['failed_users'] += 1 self.stdout.write( self.style.ERROR(f'✗ Failed to warm cache for {user.get_full_name()}: {user_results["errors"]}') ) results['errors'].extend(user_results['errors']) results['total_pages_warmed'] += user_results['pages_warmed'] results['total_permissions_cached'] += user_results['permissions_cached'] results['total_actions_cached'] += user_results['actions_cached'] except Exception as e: results['failed_users'] += 1 error_msg = f'Error warming cache for user {user.id}: {str(e)}' results['errors'].append(error_msg) self.stdout.write(self.style.ERROR(f'✗ {error_msg}')) return results def _warm_critical_users(self, limit, pages, force): """Warm cache for critical/active users""" self.stdout.write(f'Warming cache for up to {limit} critical users...') # Use the cache service's built-in critical user warming results = permission_cache.warm_critical_users_cache(limit) # Add individual user feedback if results['successful_users'] > 0: self.stdout.write( self.style.SUCCESS(f'✓ Successfully warmed cache for {results["successful_users"]} users') ) if results['failed_users'] > 0: self.stdout.write( self.style.ERROR(f'✗ Failed to warm cache for {results["failed_users"]} users') ) for error in results['errors'][:5]: # Show first 5 errors self.stdout.write(self.style.ERROR(f' - {error}')) if len(results['errors']) > 5: self.stdout.write( self.style.ERROR(f' ... and {len(results["errors"]) - 5} more errors') ) return results def _display_results(self, results, start_time): """Display warming results summary""" end_time = timezone.now() duration = (end_time - start_time).total_seconds() self.stdout.write('\n' + '='*60) self.stdout.write(self.style.SUCCESS('PERMISSION CACHE WARMING COMPLETE')) self.stdout.write('='*60) self.stdout.write(f'Duration: {duration:.2f} seconds') self.stdout.write(f'Total users processed: {results["total_users"]}') self.stdout.write(f'Successful: {results["successful_users"]}') self.stdout.write(f'Failed: {results["failed_users"]}') self.stdout.write(f'Total pages warmed: {results["total_pages_warmed"]}') self.stdout.write(f'Total permissions cached: {results["total_permissions_cached"]}') self.stdout.write(f'Total actions cached: {results["total_actions_cached"]}') if results['errors']: self.stdout.write(f'\nErrors encountered: {len(results["errors"])}') if len(results['errors']) <= 10: for error in results['errors']: self.stdout.write(self.style.ERROR(f' - {error}')) else: for error in results['errors'][:10]: self.stdout.write(self.style.ERROR(f' - {error}')) self.stdout.write(self.style.ERROR(f' ... and {len(results["errors"]) - 10} more errors')) # Display cache statistics try: stats = permission_cache.get_cache_statistics() self.stdout.write(f'\nCache Statistics:') self.stdout.write(f' Backend: {stats.get("cache_backend", "Unknown")}') self.stdout.write(f' Hit Rate: {stats.get("hit_rate_percentage", 0):.2f}%') self.stdout.write(f' Total Hits: {stats.get("total_hits", 0)}') self.stdout.write(f' Total Misses: {stats.get("total_misses", 0)}') except Exception as e: self.stdout.write(self.style.WARNING(f'Could not retrieve cache statistics: {e}')) self.stdout.write('='*60)