""" Redis-based Permission Caching System for Enhanced Performance """ import json import logging import time from typing import Dict, List, Optional, Set, Tuple, Any from django.core.cache import cache from django.conf import settings from django.utils import timezone from django.db.models.signals import post_save, post_delete from django.dispatch import receiver import redis logger = logging.getLogger(__name__) class PermissionCacheService: """ Advanced Redis-based caching service for permission system with intelligent cache warming, invalidation strategies, and performance monitoring """ # Cache key prefixes USER_PERMISSIONS_PREFIX = "perm:user" ROLE_TEMPLATE_PREFIX = "perm:role" PAGE_PERMISSIONS_PREFIX = "perm:page" BULK_CHECK_PREFIX = "perm:bulk" CACHE_STATS_PREFIX = "perm:stats" # Cache timeouts (in seconds) USER_PERMISSION_TIMEOUT = 900 # 15 minutes ROLE_TEMPLATE_TIMEOUT = 1800 # 30 minutes PAGE_PERMISSIONS_TIMEOUT = 3600 # 1 hour BULK_CHECK_TIMEOUT = 600 # 10 minutes STATS_TIMEOUT = 300 # 5 minutes # Cache warming settings CRITICAL_USERS_THRESHOLD = 50 # Users to prioritize for cache warming WARM_CACHE_BATCH_SIZE = 10 # Number of users to warm at once def __init__(self): self.redis_client = self._get_redis_client() self.cache_stats = { 'hits': 0, 'misses': 0, 'invalidations': 0, 'warm_operations': 0 } def _get_redis_client(self): """Get Redis client instance""" try: if hasattr(settings, 'CACHES') and 'default' in settings.CACHES: cache_config = settings.CACHES['default'] if 'redis' in cache_config.get('BACKEND', '').lower(): # Extract Redis connection details location = cache_config.get('LOCATION', 'redis://127.0.0.1:6379/1') return redis.from_url(location, decode_responses=True) # Fallback to default Redis connection return redis.Redis(host='127.0.0.1', port=6379, db=1, decode_responses=True) except Exception as e: logger.warning(f"Could not connect to Redis: {e}. Falling back to Django cache.") return None def get_user_page_permissions(self, user_id: str, page_name: str) -> Optional[Dict[str, bool]]: """ Get cached user permissions for a specific page Args: user_id: User ID page_name: Page name Returns: Dictionary of permissions or None if not cached """ cache_key = f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page_name}" try: # Try Redis first if available if self.redis_client: cached_data = self.redis_client.get(cache_key) if cached_data: self._record_cache_hit() return json.loads(cached_data) # Fallback to Django cache cached_data = cache.get(cache_key) if cached_data: self._record_cache_hit() return cached_data self._record_cache_miss() return None except Exception as e: logger.error(f"Error getting cached permissions for user {user_id}, page {page_name}: {e}") self._record_cache_miss() return None def set_user_page_permissions(self, user_id: str, page_name: str, permissions: Dict[str, bool], timeout: Optional[int] = None) -> bool: """ Cache user permissions for a specific page Args: user_id: User ID page_name: Page name permissions: Dictionary of permissions to cache timeout: Optional custom timeout Returns: True if successfully cached """ cache_key = f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page_name}" timeout = timeout or self.USER_PERMISSION_TIMEOUT try: # Cache in Redis if available if self.redis_client: self.redis_client.setex( cache_key, timeout, json.dumps(permissions) ) # Also cache in Django cache as fallback cache.set(cache_key, permissions, timeout) # Update cache metadata self._update_cache_metadata(user_id, page_name, 'permissions') return True except Exception as e: logger.error(f"Error caching permissions for user {user_id}, page {page_name}: {e}") return False def get_user_available_actions(self, user_id: str, page_name: str) -> Optional[List[Dict[str, Any]]]: """ Get cached available actions for a user on a page Args: user_id: User ID page_name: Page name Returns: List of available actions or None if not cached """ cache_key = f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page_name}:actions" try: if self.redis_client: cached_data = self.redis_client.get(cache_key) if cached_data: self._record_cache_hit() return json.loads(cached_data) cached_data = cache.get(cache_key) if cached_data: self._record_cache_hit() return cached_data self._record_cache_miss() return None except Exception as e: logger.error(f"Error getting cached actions for user {user_id}, page {page_name}: {e}") self._record_cache_miss() return None def set_user_available_actions(self, user_id: str, page_name: str, actions: List[Dict[str, Any]], timeout: Optional[int] = None) -> bool: """ Cache available actions for a user on a page Args: user_id: User ID page_name: Page name actions: List of available actions timeout: Optional custom timeout Returns: True if successfully cached """ cache_key = f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page_name}:actions" timeout = timeout or self.USER_PERMISSION_TIMEOUT try: if self.redis_client: self.redis_client.setex( cache_key, timeout, json.dumps(actions) ) cache.set(cache_key, actions, timeout) self._update_cache_metadata(user_id, page_name, 'actions') return True except Exception as e: logger.error(f"Error caching actions for user {user_id}, page {page_name}: {e}") return False def get_role_template(self, role: str) -> Optional[Dict[str, Dict[str, Any]]]: """ Get cached role permission template Args: role: Role name Returns: Role template dictionary or None if not cached """ cache_key = f"{self.ROLE_TEMPLATE_PREFIX}:{role}" try: if self.redis_client: cached_data = self.redis_client.get(cache_key) if cached_data: self._record_cache_hit() return json.loads(cached_data) cached_data = cache.get(cache_key) if cached_data: self._record_cache_hit() return cached_data self._record_cache_miss() return None except Exception as e: logger.error(f"Error getting cached role template for {role}: {e}") self._record_cache_miss() return None def set_role_template(self, role: str, template: Dict[str, Dict[str, Any]], timeout: Optional[int] = None) -> bool: """ Cache role permission template Args: role: Role name template: Role template dictionary timeout: Optional custom timeout Returns: True if successfully cached """ cache_key = f"{self.ROLE_TEMPLATE_PREFIX}:{role}" timeout = timeout or self.ROLE_TEMPLATE_TIMEOUT try: if self.redis_client: self.redis_client.setex( cache_key, timeout, json.dumps(template) ) cache.set(cache_key, template, timeout) return True except Exception as e: logger.error(f"Error caching role template for {role}: {e}") return False def invalidate_user_cache(self, user_id: str, page_name: Optional[str] = None) -> int: """ Invalidate cached permissions for a user Args: user_id: User ID to invalidate cache for page_name: Optional specific page, if None invalidates all pages Returns: Number of cache keys invalidated """ try: keys_to_delete = [] if page_name: # Invalidate specific page cache keys_to_delete.extend([ f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page_name}", f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page_name}:actions" ]) else: # Find all cache keys for this user if self.redis_client: pattern = f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:*" keys_to_delete = list(self.redis_client.scan_iter(match=pattern)) else: # For Django cache, we'll use a more limited approach # In production, consider using cache tags or a more sophisticated pattern common_pages = ['loans', 'clients', 'reports', 'dashboard', 'repayments'] for page in common_pages: keys_to_delete.extend([ f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page}", f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page}:actions" ]) # Delete from Redis if self.redis_client and keys_to_delete: deleted_count = self.redis_client.delete(*keys_to_delete) else: deleted_count = 0 # Delete from Django cache if keys_to_delete: cache.delete_many(keys_to_delete) deleted_count = max(deleted_count, len(keys_to_delete)) self._record_cache_invalidation() logger.info(f"Invalidated {deleted_count} cache keys for user {user_id}") return deleted_count except Exception as e: logger.error(f"Error invalidating cache for user {user_id}: {e}") return 0 def invalidate_role_cache(self, role: str) -> int: """ Invalidate cached role template and all users with this role Args: role: Role name to invalidate Returns: Number of cache keys invalidated """ try: keys_to_delete = [] # Invalidate role template role_template_key = f"{self.ROLE_TEMPLATE_PREFIX}:{role}" keys_to_delete.append(role_template_key) # Find and invalidate all users with this role from .models import CustomUser user_ids = CustomUser.objects.filter(role=role).values_list('id', flat=True) for user_id in user_ids: if self.redis_client: pattern = f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:*" user_keys = list(self.redis_client.scan_iter(match=pattern)) keys_to_delete.extend(user_keys) else: # Limited approach for Django cache common_pages = ['loans', 'clients', 'reports', 'dashboard', 'repayments'] for page in common_pages: keys_to_delete.extend([ f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page}", f"{self.USER_PERMISSIONS_PREFIX}:{user_id}:{page}:actions" ]) # Delete from Redis if self.redis_client and keys_to_delete: deleted_count = self.redis_client.delete(*keys_to_delete) else: deleted_count = 0 # Delete from Django cache if keys_to_delete: cache.delete_many(keys_to_delete) deleted_count = max(deleted_count, len(keys_to_delete)) self._record_cache_invalidation() logger.info(f"Invalidated {deleted_count} cache keys for role {role}") return deleted_count except Exception as e: logger.error(f"Error invalidating cache for role {role}: {e}") return 0 def warm_user_cache(self, user_id: str, pages: Optional[List[str]] = None) -> Dict[str, Any]: """ Pre-warm cache for a user's critical permissions Args: user_id: User ID to warm cache for pages: Optional list of pages to warm, if None warms common pages Returns: Dictionary with warming results """ try: from .services import PagePermissionManager from .models import CustomUser user = CustomUser.objects.get(id=user_id) permission_manager = PagePermissionManager() pages = pages or ['loans', 'clients', 'reports', 'dashboard', 'repayments'] results = { 'user_id': user_id, 'pages_warmed': 0, 'permissions_cached': 0, 'actions_cached': 0, 'errors': [] } for page_name in pages: try: # Warm page permissions permissions = permission_manager.get_page_permissions(user, page_name) if permissions: self.set_user_page_permissions(user_id, page_name, permissions) results['permissions_cached'] += len(permissions) # Warm available actions actions = permission_manager.get_available_actions(user, page_name) if actions: self.set_user_available_actions(user_id, page_name, actions) results['actions_cached'] += len(actions) results['pages_warmed'] += 1 except Exception as e: error_msg = f"Error warming cache for page {page_name}: {str(e)}" results['errors'].append(error_msg) logger.error(error_msg) self._record_cache_warm_operation() logger.info(f"Warmed cache for user {user_id}: {results['pages_warmed']} pages, " f"{results['permissions_cached']} permissions, {results['actions_cached']} actions") return results except Exception as e: logger.error(f"Error warming cache for user {user_id}: {e}") return { 'user_id': user_id, 'pages_warmed': 0, 'permissions_cached': 0, 'actions_cached': 0, 'errors': [str(e)] } def warm_critical_users_cache(self, limit: Optional[int] = None) -> Dict[str, Any]: """ Warm cache for critical/active users Args: limit: Optional limit on number of users to warm Returns: Dictionary with warming results """ try: from .models import CustomUser from django.utils import timezone from datetime import timedelta # Get recently active users or users with critical roles recent_threshold = timezone.now() - timedelta(days=7) critical_roles = ['admin', 'team_leader', 'loan_officer'] users_query = CustomUser.objects.filter( Q(last_login__gte=recent_threshold) | Q(role__in=critical_roles), is_active=True ).order_by('-last_login') if limit: users_query = users_query[:limit] else: users_query = users_query[:self.CRITICAL_USERS_THRESHOLD] 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 in users_query: try: user_results = self.warm_user_cache(str(user.id)) results['total_users'] += 1 if not user_results.get('errors'): results['successful_users'] += 1 else: results['failed_users'] += 1 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) logger.error(error_msg) logger.info(f"Warmed cache for {results['successful_users']}/{results['total_users']} critical users") return results except Exception as e: logger.error(f"Error warming critical users cache: {e}") return { 'total_users': 0, 'successful_users': 0, 'failed_users': 1, 'errors': [str(e)] } def get_cache_statistics(self) -> Dict[str, Any]: """ Get cache performance statistics Returns: Dictionary with cache statistics """ try: stats_key = f"{self.CACHE_STATS_PREFIX}:performance" # Get current stats from cache if self.redis_client: cached_stats = self.redis_client.get(stats_key) if cached_stats: persistent_stats = json.loads(cached_stats) else: persistent_stats = { 'total_hits': 0, 'total_misses': 0, 'total_invalidations': 0, 'total_warm_operations': 0 } else: persistent_stats = cache.get(stats_key, { 'total_hits': 0, 'total_misses': 0, 'total_invalidations': 0, 'total_warm_operations': 0 }) # Combine with current session stats total_requests = persistent_stats['total_hits'] + persistent_stats['total_misses'] hit_rate = (persistent_stats['total_hits'] / total_requests * 100) if total_requests > 0 else 0 statistics = { 'cache_backend': 'Redis' if self.redis_client else 'Django Database', 'total_hits': persistent_stats['total_hits'] + self.cache_stats['hits'], 'total_misses': persistent_stats['total_misses'] + self.cache_stats['misses'], 'total_invalidations': persistent_stats['total_invalidations'] + self.cache_stats['invalidations'], 'total_warm_operations': persistent_stats['total_warm_operations'] + self.cache_stats['warm_operations'], 'hit_rate_percentage': round(hit_rate, 2), 'session_stats': self.cache_stats.copy(), 'last_updated': timezone.now().isoformat() } # Get Redis-specific stats if available if self.redis_client: try: redis_info = self.redis_client.info() statistics['redis_stats'] = { 'used_memory': redis_info.get('used_memory_human', 'N/A'), 'connected_clients': redis_info.get('connected_clients', 0), 'total_commands_processed': redis_info.get('total_commands_processed', 0), 'keyspace_hits': redis_info.get('keyspace_hits', 0), 'keyspace_misses': redis_info.get('keyspace_misses', 0) } except Exception as e: logger.warning(f"Could not get Redis stats: {e}") return statistics except Exception as e: logger.error(f"Error getting cache statistics: {e}") return { 'error': str(e), 'cache_backend': 'Unknown', 'total_hits': 0, 'total_misses': 0, 'hit_rate_percentage': 0 } def _record_cache_hit(self): """Record a cache hit for statistics""" self.cache_stats['hits'] += 1 self._update_persistent_stats('hits') def _record_cache_miss(self): """Record a cache miss for statistics""" self.cache_stats['misses'] += 1 self._update_persistent_stats('misses') def _record_cache_invalidation(self): """Record a cache invalidation for statistics""" self.cache_stats['invalidations'] += 1 self._update_persistent_stats('invalidations') def _record_cache_warm_operation(self): """Record a cache warm operation for statistics""" self.cache_stats['warm_operations'] += 1 self._update_persistent_stats('warm_operations') def _update_persistent_stats(self, stat_type: str): """Update persistent cache statistics""" try: stats_key = f"{self.CACHE_STATS_PREFIX}:performance" if self.redis_client: # Use Redis atomic increment field_key = f"total_{stat_type}" self.redis_client.hincrby(stats_key, field_key, 1) self.redis_client.expire(stats_key, self.STATS_TIMEOUT * 12) # Keep stats longer except Exception as e: logger.debug(f"Error updating persistent stats: {e}") def _update_cache_metadata(self, user_id: str, page_name: str, cache_type: str): """Update cache metadata for monitoring""" try: metadata_key = f"{self.CACHE_STATS_PREFIX}:metadata:{user_id}" metadata = { 'last_cached': timezone.now().isoformat(), 'page_name': page_name, 'cache_type': cache_type } if self.redis_client: self.redis_client.setex( metadata_key, self.STATS_TIMEOUT, json.dumps(metadata) ) except Exception as e: logger.debug(f"Error updating cache metadata: {e}") # Global cache service instance permission_cache = PermissionCacheService() # Signal handlers for automatic cache invalidation @receiver(post_save, sender='users.UserPagePermission') def invalidate_user_permission_cache(sender, instance, **kwargs): """Invalidate user cache when permissions are updated""" try: permission_cache.invalidate_user_cache(str(instance.user.id)) logger.info(f"Invalidated cache for user {instance.user.id} due to permission update") except Exception as e: logger.error(f"Error invalidating user cache on permission update: {e}") @receiver(post_save, sender='users.RolePermissionTemplate') def invalidate_role_template_cache(sender, instance, **kwargs): """Invalidate role cache when role templates are updated""" try: permission_cache.invalidate_role_cache(instance.role) logger.info(f"Invalidated cache for role {instance.role} due to template update") except Exception as e: logger.error(f"Error invalidating role cache on template update: {e}") @receiver(post_delete, sender='users.UserPagePermission') def invalidate_user_permission_cache_on_delete(sender, instance, **kwargs): """Invalidate user cache when permissions are deleted""" try: permission_cache.invalidate_user_cache(str(instance.user.id)) logger.info(f"Invalidated cache for user {instance.user.id} due to permission deletion") except Exception as e: logger.error(f"Error invalidating user cache on permission deletion: {e}")