""" Management command to optimize permission system database performance """ from django.core.management.base import BaseCommand, CommandError from django.utils import timezone from django.db import connection from users.database_optimization_service import db_optimizer import logging logger = logging.getLogger(__name__) class Command(BaseCommand): help = 'Optimize permission system database performance with indexes and query optimization' def add_arguments(self, parser): parser.add_argument( '--create-indexes', action='store_true', help='Create optimized database indexes' ) parser.add_argument( '--optimize-queries', action='store_true', help='Optimize and cache common queries' ) parser.add_argument( '--analyze-performance', action='store_true', help='Analyze database performance and generate recommendations' ) parser.add_argument( '--full-optimization', action='store_true', help='Run complete database optimization (all options)' ) parser.add_argument( '--vacuum-analyze', action='store_true', help='Run VACUUM ANALYZE on permission tables' ) parser.add_argument( '--connection-pool-size', type=int, default=10, help='Recommended connection pool size (default: 10)' ) def handle(self, *args, **options): start_time = timezone.now() self.stdout.write( self.style.SUCCESS(f'Starting database optimization at {start_time}') ) try: if options['full_optimization']: # Run all optimizations self._run_full_optimization(options) else: # Run specific optimizations if options['create_indexes']: self._create_indexes() if options['optimize_queries']: self._optimize_queries() if options['analyze_performance']: self._analyze_performance() if options['vacuum_analyze']: self._vacuum_analyze_tables() # Always show performance statistics self._show_performance_statistics() # Show connection pool recommendations if options['connection_pool_size']: self._show_connection_pool_config(options['connection_pool_size']) end_time = timezone.now() duration = (end_time - start_time).total_seconds() self.stdout.write( self.style.SUCCESS(f'Database optimization completed in {duration:.2f} seconds') ) except Exception as e: raise CommandError(f'Error during database optimization: {e}') def _run_full_optimization(self, options): """Run complete database optimization""" self.stdout.write(self.style.WARNING('Running full database optimization...')) # Step 1: Create indexes self.stdout.write('Step 1: Creating optimized indexes...') self._create_indexes() # Step 2: Optimize queries self.stdout.write('Step 2: Optimizing queries and caching...') self._optimize_queries() # Step 3: Vacuum and analyze self.stdout.write('Step 3: Running VACUUM ANALYZE...') self._vacuum_analyze_tables() # Step 4: Performance analysis self.stdout.write('Step 4: Analyzing performance...') self._analyze_performance() self.stdout.write(self.style.SUCCESS('✓ Full optimization completed')) def _create_indexes(self): """Create optimized database indexes""" self.stdout.write('Creating optimized database indexes...') try: # Use the optimization service to create indexes results = db_optimizer.optimize_permission_queries() if 'error' in results: self.stdout.write( self.style.ERROR(f'Error creating indexes: {results["error"]}') ) else: self.stdout.write( self.style.SUCCESS(f'✓ Created {results["indexes_created"]} index groups') ) if results.get('performance_improvements'): for improvement in results['performance_improvements']: self.stdout.write(f' - {improvement}') except Exception as e: self.stdout.write( self.style.ERROR(f'Error creating indexes: {e}') ) def _optimize_queries(self): """Optimize common queries and enable caching""" self.stdout.write('Optimizing queries and enabling caching...') try: # Optimize role template queries db_optimizer._optimize_role_template_queries() self.stdout.write(' ✓ Optimized role template queries') # Optimize user permission queries db_optimizer._optimize_user_permission_queries() self.stdout.write(' ✓ Optimized user permission queries') # Optimize analytics queries db_optimizer._optimize_analytics_queries() self.stdout.write(' ✓ Optimized analytics queries') self.stdout.write(self.style.SUCCESS('✓ Query optimization completed')) except Exception as e: self.stdout.write( self.style.ERROR(f'Error optimizing queries: {e}') ) def _vacuum_analyze_tables(self): """Run VACUUM ANALYZE on permission tables""" self.stdout.write('Running VACUUM ANALYZE on permission tables...') permission_tables = [ 'page_permissions', 'role_permission_templates', 'user_page_permissions', 'portfolio_snapshots', 'client_growth_metrics' ] try: with connection.cursor() as cursor: for table in permission_tables: try: self.stdout.write(f' Analyzing {table}...') cursor.execute(f'VACUUM ANALYZE {table}') self.stdout.write(f' ✓ {table} analyzed') except Exception as e: self.stdout.write( self.style.WARNING(f' Could not analyze {table}: {e}') ) self.stdout.write(self.style.SUCCESS('✓ VACUUM ANALYZE completed')) except Exception as e: self.stdout.write( self.style.ERROR(f'Error running VACUUM ANALYZE: {e}') ) def _analyze_performance(self): """Analyze database performance and show recommendations""" self.stdout.write('Analyzing database performance...') try: # Generate performance recommendations recommendations = db_optimizer._generate_performance_recommendations() if recommendations: self.stdout.write('\n' + '='*60) self.stdout.write(self.style.SUCCESS('PERFORMANCE RECOMMENDATIONS')) self.stdout.write('='*60) for rec in recommendations: priority_color = { 'high': self.style.ERROR, 'medium': self.style.WARNING, 'low': self.style.SUCCESS }.get(rec.get('priority', 'low'), self.style.SUCCESS) self.stdout.write( priority_color(f"\n[{rec.get('priority', 'LOW').upper()}] {rec.get('title', 'Unknown')}") ) self.stdout.write(f" {rec.get('description', 'No description')}") if rec.get('details'): if isinstance(rec['details'], dict): for key, value in rec['details'].items(): self.stdout.write(f" {key}: {value}") elif isinstance(rec['details'], list): for item in rec['details'][:3]: # Show first 3 items self.stdout.write(f" - {item}") if len(rec['details']) > 3: self.stdout.write(f" ... and {len(rec['details']) - 3} more") self.stdout.write('='*60) else: self.stdout.write(self.style.SUCCESS('✓ No specific performance issues detected')) except Exception as e: self.stdout.write( self.style.ERROR(f'Error analyzing performance: {e}') ) def _show_performance_statistics(self): """Show current performance statistics""" self.stdout.write('\n' + '='*60) self.stdout.write(self.style.SUCCESS('DATABASE PERFORMANCE STATISTICS')) self.stdout.write('='*60) try: stats = db_optimizer.get_performance_statistics() # Optimization stats opt_stats = stats.get('optimization_stats', {}) self.stdout.write(f"Optimized Queries: {opt_stats.get('optimized_queries', 0)}") self.stdout.write(f"Cache Hits: {opt_stats.get('cache_hits', 0)}") self.stdout.write(f"Index Recommendations: {opt_stats.get('index_recommendations', 0)}") # Cache info cache_info = stats.get('cache_info', {}) self.stdout.write(f"Query Cache Timeout: {cache_info.get('query_cache_timeout', 0)} seconds") self.stdout.write(f"Cache Backend: {cache_info.get('cache_backend', 'Unknown')}") # Database info db_info = stats.get('database_info', {}) if 'version' in db_info: version_short = db_info['version'].split(' ')[0:2] self.stdout.write(f"Database: {' '.join(version_short)}") if 'active_connections' in db_info: self.stdout.write(f"Active Connections: {db_info['active_connections']}") if 'cache_hit_ratio' in db_info: hit_ratio = db_info['cache_hit_ratio'] if hit_ratio >= 95: color = self.style.SUCCESS elif hit_ratio >= 85: color = self.style.WARNING else: color = self.style.ERROR self.stdout.write(color(f"Database Cache Hit Ratio: {hit_ratio}%")) if 'error' in db_info: self.stdout.write( self.style.WARNING(f"Database info error: {db_info['error']}") ) except Exception as e: self.stdout.write( self.style.ERROR(f'Error getting performance statistics: {e}') ) self.stdout.write('='*60) def _show_connection_pool_config(self, pool_size): """Show connection pool configuration recommendations""" self.stdout.write('\n' + '='*60) self.stdout.write(self.style.SUCCESS('CONNECTION POOL RECOMMENDATIONS')) self.stdout.write('='*60) try: pool_config = db_optimizer.create_connection_pool(pool_size) if pool_config: self.stdout.write('Add the following to your Django settings.py:') self.stdout.write('') self.stdout.write('DATABASES = {') self.stdout.write(' "default": {') self.stdout.write(f' "ENGINE": "{pool_config["ENGINE"]}",') self.stdout.write(' "NAME": "your_database_name",') self.stdout.write(' "USER": "your_database_user",') self.stdout.write(' "PASSWORD": "your_database_password",') self.stdout.write(' "HOST": "your_database_host",') self.stdout.write(' "PORT": "your_database_port",') self.stdout.write(' "OPTIONS": {') for key, value in pool_config['OPTIONS'].items(): self.stdout.write(f' "{key}": {value},') self.stdout.write(' },') if pool_config.get('CONN_HEALTH_CHECKS'): self.stdout.write(f' "CONN_HEALTH_CHECKS": {pool_config["CONN_HEALTH_CHECKS"]},') self.stdout.write(' }') self.stdout.write('}') self.stdout.write('') self.stdout.write('Consider also installing django-db-pool or similar for advanced pooling.') except Exception as e: self.stdout.write( self.style.ERROR(f'Error generating connection pool config: {e}') ) self.stdout.write('='*60) def _test_query_performance(self): """Test query performance with sample operations""" self.stdout.write('\n' + '='*60) self.stdout.write(self.style.SUCCESS('QUERY PERFORMANCE TEST')) self.stdout.write('='*60) try: import time from users.models import CustomUser # Test user permission lookup start_time = time.time() users = CustomUser.objects.filter(is_active=True)[:10] for user in users: # Test optimized permission check permissions = db_optimizer.get_optimized_user_permissions(str(user.id), 'loans') end_time = time.time() permission_time = (end_time - start_time) * 1000 # Convert to milliseconds self.stdout.write(f"Permission lookup (10 users): {permission_time:.2f}ms") # Test role template lookup start_time = time.time() roles = ['admin', 'team_leader', 'loan_officer', 'secretary', 'auditor'] for role in roles: template = db_optimizer.get_optimized_role_template(role) end_time = time.time() template_time = (end_time - start_time) * 1000 self.stdout.write(f"Role template lookup (5 roles): {template_time:.2f}ms") # Test bulk permission check start_time = time.time() if users: test_permissions = [ ('loans', 'create_application'), ('loans', 'approve_application'), ('clients', 'view_list'), ('reports', 'export_pdf') ] bulk_results = db_optimizer.bulk_permission_check(str(users[0].id), test_permissions) end_time = time.time() bulk_time = (end_time - start_time) * 1000 self.stdout.write(f"Bulk permission check (4 permissions): {bulk_time:.2f}ms") # Performance assessment total_time = permission_time + template_time + bulk_time if total_time < 100: self.stdout.write(self.style.SUCCESS(f"✓ Excellent performance: {total_time:.2f}ms total")) elif total_time < 500: self.stdout.write(self.style.WARNING(f"⚠ Good performance: {total_time:.2f}ms total")) else: self.stdout.write(self.style.ERROR(f"✗ Poor performance: {total_time:.2f}ms total")) except Exception as e: self.stdout.write( self.style.ERROR(f'Error testing query performance: {e}') ) self.stdout.write('='*60)