""" Database Optimization Service for Permission System Performance """ import logging from typing import Dict, List, Optional, Any, Tuple from django.db import connection, transaction from django.db.models import Q, Count, Avg, Max, Min, F, Prefetch from django.core.cache import cache from django.utils import timezone from django.conf import settings import time logger = logging.getLogger(__name__) class DatabaseOptimizationService: """ Service for optimizing database queries and managing indexes for the permission system """ def __init__(self): self.query_cache_timeout = 300 # 5 minutes self.performance_stats = { 'optimized_queries': 0, 'cache_hits': 0, 'index_recommendations': 0 } def optimize_permission_queries(self): """ Optimize common permission-related database queries """ optimizations = { 'indexes_created': 0, 'queries_optimized': 0, 'performance_improvements': [], 'recommendations': [] } try: # Create optimized indexes for permission lookups self._create_permission_indexes() optimizations['indexes_created'] += 1 # Optimize role template queries self._optimize_role_template_queries() optimizations['queries_optimized'] += 1 # Optimize user permission queries self._optimize_user_permission_queries() optimizations['queries_optimized'] += 1 # Optimize analytics queries self._optimize_analytics_queries() optimizations['queries_optimized'] += 1 # Generate performance recommendations recommendations = self._generate_performance_recommendations() optimizations['recommendations'] = recommendations logger.info(f"Database optimization completed: {optimizations}") return optimizations except Exception as e: logger.error(f"Error in database optimization: {e}") optimizations['error'] = str(e) return optimizations def _create_permission_indexes(self): """ Create optimized database indexes for permission system """ with connection.cursor() as cursor: # Composite indexes for permission lookups permission_indexes = [ # PagePermission indexes "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_page_perm_lookup ON page_permissions (page_name, action_code, is_active)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_page_perm_category ON page_permissions (page_name, category, is_active)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_page_perm_critical ON page_permissions (is_critical, is_active)", # RolePermissionTemplate indexes "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_role_template_lookup ON role_permission_templates (role, is_allowed)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_role_template_override ON role_permission_templates (role, can_override, is_allowed)", # UserPagePermission indexes "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_perm_lookup ON user_page_permissions (user_id, is_allowed)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_perm_expiry ON user_page_permissions (user_id, expires_at) WHERE expires_at IS NOT NULL", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_user_perm_granted ON user_page_permissions (granted_by_id, created_at)", # Portfolio analytics indexes "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_portfolio_manager_date ON portfolio_snapshots (manager_id, snapshot_date DESC)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_portfolio_branch_date ON portfolio_snapshots (branch_id, snapshot_date DESC)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_portfolio_performance ON portfolio_snapshots (collection_rate, default_rate, snapshot_date)", # Client growth metrics indexes "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_growth_branch_period ON client_growth_metrics (branch_id, period_type, period_end DESC)", "CREATE INDEX CONCURRENTLY IF NOT EXISTS idx_growth_performance ON client_growth_metrics (new_clients, churned_clients, period_end)", ] for index_sql in permission_indexes: try: cursor.execute(index_sql) logger.info(f"Created index: {index_sql.split('idx_')[1].split(' ')[0]}") except Exception as e: if "already exists" not in str(e).lower(): logger.warning(f"Could not create index: {e}") def _optimize_role_template_queries(self): """ Optimize role template related queries with prefetching and caching """ from .enhanced_permissions_models import RolePermissionTemplate # Pre-warm role template cache with optimized queries roles = ['admin', 'team_leader', 'loan_officer', 'secretary', 'auditor'] for role in roles: cache_key = f"optimized_role_template:{role}" if not cache.get(cache_key): # Use optimized query with select_related and prefetch_related templates = RolePermissionTemplate.objects.filter( role=role ).select_related( 'page_permission', 'created_by' ).prefetch_related( 'page_permission__required_permissions' ).order_by( 'page_permission__page_name', 'page_permission__category', 'page_permission__action_name' ) # Convert to dictionary for caching template_dict = {} for template in templates: page_name = template.page_permission.page_name action_code = template.page_permission.action_code if page_name not in template_dict: template_dict[page_name] = {} template_dict[page_name][action_code] = { 'is_allowed': template.is_allowed, 'can_override': template.can_override, 'priority': template.priority, 'action_name': template.page_permission.action_name, 'description': template.page_permission.description, 'category': template.page_permission.category, 'is_critical': template.page_permission.is_critical, } cache.set(cache_key, template_dict, self.query_cache_timeout) logger.debug(f"Cached optimized role template for {role}") def _optimize_user_permission_queries(self): """ Optimize user permission queries with bulk operations and caching """ from .enhanced_permissions_models import UserPagePermission from .models import CustomUser # Get frequently accessed users (recent logins) recent_threshold = timezone.now() - timezone.timedelta(days=7) active_users = CustomUser.objects.filter( last_login__gte=recent_threshold, is_active=True ).select_related('branch').order_by('-last_login')[:50] # Bulk prefetch user permissions user_permissions = UserPagePermission.objects.filter( user__in=active_users ).select_related( 'user', 'page_permission', 'granted_by' ).prefetch_related( 'page_permission__required_permissions' ) # Group by user for efficient caching user_permission_map = {} for perm in user_permissions: user_id = str(perm.user.id) if user_id not in user_permission_map: user_permission_map[user_id] = {} page_name = perm.page_permission.page_name if page_name not in user_permission_map[user_id]: user_permission_map[user_id][page_name] = {} user_permission_map[user_id][page_name][perm.page_permission.action_code] = { 'is_allowed': perm.is_allowed, 'expires_at': perm.expires_at.isoformat() if perm.expires_at else None, 'granted_by': perm.granted_by.get_full_name() if perm.granted_by else None, 'reason': perm.reason } # Cache user permissions for user_id, permissions in user_permission_map.items(): cache_key = f"optimized_user_permissions:{user_id}" cache.set(cache_key, permissions, self.query_cache_timeout) logger.info(f"Optimized permissions for {len(user_permission_map)} active users") def _optimize_analytics_queries(self): """ Optimize analytics queries with aggregation and materialized views """ from .enhanced_permissions_models import PortfolioSnapshot, ClientGrowthMetrics # Create optimized analytics cache analytics_cache = {} # Portfolio performance aggregations portfolio_stats = PortfolioSnapshot.objects.aggregate( avg_collection_rate=Avg('collection_rate'), avg_default_rate=Avg('default_rate'), avg_portfolio_yield=Avg('portfolio_yield'), total_outstanding=Sum('total_outstanding'), total_disbursed=Sum('total_disbursed') ) analytics_cache['portfolio_overview'] = portfolio_stats # Recent portfolio trends (last 30 days) recent_date = timezone.now().date() - timezone.timedelta(days=30) recent_snapshots = PortfolioSnapshot.objects.filter( snapshot_date__gte=recent_date ).values('snapshot_date').annotate( total_clients=Sum('total_clients'), total_outstanding=Sum('total_outstanding'), avg_collection_rate=Avg('collection_rate') ).order_by('snapshot_date') analytics_cache['recent_trends'] = list(recent_snapshots) # Top performing managers top_managers = PortfolioSnapshot.objects.filter( snapshot_date__gte=recent_date ).values( 'manager__first_name', 'manager__last_name', 'manager_id' ).annotate( avg_collection_rate=Avg('collection_rate'), total_clients=Avg('total_clients'), total_outstanding=Avg('total_outstanding') ).order_by('-avg_collection_rate')[:10] analytics_cache['top_managers'] = list(top_managers) # Client growth trends growth_trends = ClientGrowthMetrics.objects.filter( period_end__gte=recent_date ).values('period_end').annotate( total_new_clients=Sum('new_clients'), total_churned_clients=Sum('churned_clients'), avg_acquisition_cost=Avg('acquisition_cost') ).order_by('period_end') analytics_cache['growth_trends'] = list(growth_trends) # Cache all analytics data cache.set('optimized_analytics_overview', analytics_cache, self.query_cache_timeout * 2) logger.info("Cached optimized analytics data") def get_optimized_user_permissions(self, user_id: str, page_name: str) -> Optional[Dict]: """ Get user permissions using optimized queries and caching """ cache_key = f"optimized_user_permissions:{user_id}" cached_permissions = cache.get(cache_key) if cached_permissions and page_name in cached_permissions: self.performance_stats['cache_hits'] += 1 return cached_permissions[page_name] # Fallback to database with optimized query from .enhanced_permissions_models import UserPagePermission permissions = UserPagePermission.objects.filter( user_id=user_id, page_permission__page_name=page_name, page_permission__is_active=True ).select_related( 'page_permission', 'granted_by' ).values( 'page_permission__action_code', 'is_allowed', 'expires_at', 'granted_by__first_name', 'granted_by__last_name', 'reason' ) permission_dict = {} for perm in permissions: action_code = perm['page_permission__action_code'] permission_dict[action_code] = { 'is_allowed': perm['is_allowed'], 'expires_at': perm['expires_at'].isoformat() if perm['expires_at'] else None, 'granted_by': f"{perm['granted_by__first_name']} {perm['granted_by__last_name']}" if perm['granted_by__first_name'] else None, 'reason': perm['reason'] } return permission_dict def get_optimized_role_template(self, role: str) -> Optional[Dict]: """ Get role template using optimized queries and caching """ cache_key = f"optimized_role_template:{role}" cached_template = cache.get(cache_key) if cached_template: self.performance_stats['cache_hits'] += 1 return cached_template # Fallback to database optimization self._optimize_role_template_queries() return cache.get(cache_key) def get_optimized_analytics_data(self, data_type: str) -> Optional[Any]: """ Get analytics data using optimized queries and caching """ analytics_cache = cache.get('optimized_analytics_overview', {}) if data_type in analytics_cache: self.performance_stats['cache_hits'] += 1 return analytics_cache[data_type] # Refresh analytics cache if not found self._optimize_analytics_queries() analytics_cache = cache.get('optimized_analytics_overview', {}) return analytics_cache.get(data_type) def bulk_permission_check(self, user_id: str, permission_checks: List[Tuple[str, str]]) -> Dict[str, bool]: """ Efficiently check multiple permissions at once using optimized queries """ from .enhanced_permissions_models import PagePermission, UserPagePermission, RolePermissionTemplate from .models import CustomUser try: user = CustomUser.objects.select_related('branch').get(id=user_id) # Admin users have all permissions if user.role == 'admin': return {f"{page}:{action}": True for page, action in permission_checks} results = {} # Get all relevant page permissions in one query page_names = list(set(page for page, action in permission_checks)) action_codes = list(set(action for page, action in permission_checks)) page_permissions = PagePermission.objects.filter( page_name__in=page_names, action_code__in=action_codes, is_active=True ).select_related().prefetch_related('required_permissions') # Create lookup dictionary perm_lookup = { (perm.page_name, perm.action_code): perm for perm in page_permissions } # Get user custom permissions in bulk user_permissions = UserPagePermission.objects.filter( user=user, page_permission__in=page_permissions ).select_related('page_permission') user_perm_lookup = { (perm.page_permission.page_name, perm.page_permission.action_code): perm for perm in user_permissions } # Get role template permissions in bulk role_templates = RolePermissionTemplate.objects.filter( role=user.role, page_permission__in=page_permissions ).select_related('page_permission') role_perm_lookup = { (template.page_permission.page_name, template.page_permission.action_code): template for template in role_templates } # Check each permission for page_name, action_code in permission_checks: key = f"{page_name}:{action_code}" # Check if permission exists if (page_name, action_code) not in perm_lookup: results[key] = False continue # Check custom user permission first if (page_name, action_code) in user_perm_lookup: user_perm = user_perm_lookup[(page_name, action_code)] if not user_perm.is_expired(): results[key] = user_perm.is_allowed continue # Check role template permission if (page_name, action_code) in role_perm_lookup: role_template = role_perm_lookup[(page_name, action_code)] results[key] = role_template.is_allowed else: results[key] = False return results except Exception as e: logger.error(f"Error in bulk permission check: {e}") return {f"{page}:{action}": False for page, action in permission_checks} def _generate_performance_recommendations(self) -> List[Dict[str, Any]]: """ Generate database performance recommendations based on query analysis """ recommendations = [] try: with connection.cursor() as cursor: # Check for missing indexes cursor.execute(""" SELECT schemaname, tablename, attname, n_distinct, correlation FROM pg_stats WHERE schemaname = 'public' AND tablename IN ('page_permissions', 'role_permission_templates', 'user_page_permissions') AND n_distinct > 100 ORDER BY n_distinct DESC """) high_cardinality_columns = cursor.fetchall() if high_cardinality_columns: recommendations.append({ 'type': 'index', 'priority': 'high', 'title': 'High Cardinality Columns Detected', 'description': 'Consider adding indexes to high cardinality columns for better query performance', 'details': high_cardinality_columns[:5] }) # Check for slow queries (if pg_stat_statements is available) try: cursor.execute(""" SELECT query, calls, total_time, mean_time FROM pg_stat_statements WHERE query LIKE '%permission%' AND mean_time > 100 ORDER BY mean_time DESC LIMIT 5 """) slow_queries = cursor.fetchall() if slow_queries: recommendations.append({ 'type': 'query', 'priority': 'medium', 'title': 'Slow Permission Queries Detected', 'description': 'Some permission-related queries are taking longer than expected', 'details': [{'query': q[0][:100], 'mean_time': q[3]} for q in slow_queries] }) except Exception: # pg_stat_statements not available pass # Check table sizes cursor.execute(""" SELECT schemaname, tablename, pg_size_pretty(pg_total_relation_size(schemaname||'.'||tablename)) as size, pg_total_relation_size(schemaname||'.'||tablename) as size_bytes FROM pg_tables WHERE schemaname = 'public' AND tablename IN ('page_permissions', 'role_permission_templates', 'user_page_permissions', 'portfolio_snapshots') ORDER BY size_bytes DESC """) table_sizes = cursor.fetchall() for table in table_sizes: if table[3] > 100 * 1024 * 1024: # > 100MB recommendations.append({ 'type': 'maintenance', 'priority': 'low', 'title': f'Large Table: {table[1]}', 'description': f'Table {table[1]} is {table[2]}. Consider archiving old data or partitioning.', 'details': {'table': table[1], 'size': table[2]} }) except Exception as e: logger.warning(f"Could not generate all performance recommendations: {e}") recommendations.append({ 'type': 'error', 'priority': 'low', 'title': 'Performance Analysis Limited', 'description': 'Could not perform complete performance analysis. Some database features may not be available.', 'details': {'error': str(e)} }) # Add general recommendations recommendations.extend([ { 'type': 'cache', 'priority': 'medium', 'title': 'Enable Query Result Caching', 'description': 'Consider implementing query result caching for frequently accessed permission data', 'details': {'cache_timeout': self.query_cache_timeout} }, { 'type': 'connection', 'priority': 'low', 'title': 'Database Connection Pooling', 'description': 'Implement connection pooling for better database performance under load', 'details': {'recommended_pool_size': '10-20'} } ]) return recommendations def get_performance_statistics(self) -> Dict[str, Any]: """ Get database performance statistics """ stats = { 'optimization_stats': self.performance_stats.copy(), 'cache_info': { 'query_cache_timeout': self.query_cache_timeout, 'cache_backend': getattr(settings, 'CACHES', {}).get('default', {}).get('BACKEND', 'Unknown') }, 'database_info': {}, 'recommendations_count': 0 } try: with connection.cursor() as cursor: # Get database version and stats cursor.execute("SELECT version()") db_version = cursor.fetchone()[0] stats['database_info']['version'] = db_version # Get connection info cursor.execute("SELECT count(*) FROM pg_stat_activity") active_connections = cursor.fetchone()[0] stats['database_info']['active_connections'] = active_connections # Get cache hit ratio cursor.execute(""" SELECT sum(heap_blks_hit) / (sum(heap_blks_hit) + sum(heap_blks_read)) * 100 as cache_hit_ratio FROM pg_statio_user_tables WHERE schemaname = 'public' """) cache_hit_ratio = cursor.fetchone()[0] if cache_hit_ratio: stats['database_info']['cache_hit_ratio'] = round(float(cache_hit_ratio), 2) except Exception as e: logger.warning(f"Could not get database statistics: {e}") stats['database_info']['error'] = str(e) return stats def create_connection_pool(self, pool_size: int = 10): """ Create database connection pool for better performance """ try: # This would typically be configured in Django settings # Here we provide recommendations for connection pooling pool_config = { 'ENGINE': 'django.db.backends.postgresql', 'OPTIONS': { 'MAX_CONNS': pool_size, 'MIN_CONNS': 2, 'CONN_MAX_AGE': 600, # 10 minutes }, 'CONN_HEALTH_CHECKS': True, } logger.info(f"Connection pool configuration recommended: {pool_config}") return pool_config except Exception as e: logger.error(f"Error creating connection pool configuration: {e}") return None # Global optimization service instance db_optimizer = DatabaseOptimizationService() # Import here to avoid circular imports from django.db.models import Sum