""" Feature flags for gradual migration between permission systems """ from django.conf import settings from django.core.cache import cache from django.utils import timezone from typing import Dict, Any, Optional import logging logger = logging.getLogger(__name__) class PermissionFeatureFlags: """ Manages feature flags for permission system migration """ # Default feature flag values DEFAULT_FLAGS = { 'use_granular_permissions': False, 'fallback_to_legacy': True, 'enable_permission_caching': True, 'enable_permission_audit': True, 'enable_role_templates': False, 'enable_page_permissions': False, 'enable_permission_dependencies': False, 'enable_permission_expiry': False, 'enable_critical_permissions': False, 'enable_bulk_permission_updates': False, 'enable_permission_analytics': False, 'enable_permission_validation': True, 'migration_mode': 'disabled', # disabled, testing, gradual, complete 'debug_permissions': False, } # Cache timeout for feature flags (in seconds) CACHE_TIMEOUT = 300 # 5 minutes def __init__(self): self.cache_prefix = 'permission_flags_' def is_enabled(self, flag_name: str, user=None, default=None) -> bool: """ Check if a feature flag is enabled Args: flag_name: Name of the feature flag user: Optional user for user-specific flags default: Default value if flag not found Returns: bool: True if flag is enabled """ # Check user-specific flag first if user: user_flag = self._get_user_flag(flag_name, user) if user_flag is not None: return user_flag # Check global flag global_flag = self._get_global_flag(flag_name) if global_flag is not None: return global_flag # Use provided default or system default if default is not None: return default return self.DEFAULT_FLAGS.get(flag_name, False) def set_flag(self, flag_name: str, value: bool, user=None, expires_in: Optional[int] = None): """ Set a feature flag value Args: flag_name: Name of the feature flag value: Flag value user: Optional user for user-specific flags expires_in: Optional expiration time in seconds """ if user: self._set_user_flag(flag_name, user, value, expires_in) else: self._set_global_flag(flag_name, value, expires_in) def get_all_flags(self, user=None) -> Dict[str, Any]: """ Get all feature flags and their values Args: user: Optional user for user-specific flags Returns: Dict of flag names and values """ flags = {} # Start with defaults flags.update(self.DEFAULT_FLAGS) # Override with global flags global_flags = self._get_all_global_flags() flags.update(global_flags) # Override with user-specific flags if user: user_flags = self._get_all_user_flags(user) flags.update(user_flags) return flags def clear_flag(self, flag_name: str, user=None): """ Clear a feature flag (revert to default) Args: flag_name: Name of the feature flag user: Optional user for user-specific flags """ if user: self._clear_user_flag(flag_name, user) else: self._clear_global_flag(flag_name) def clear_all_flags(self, user=None): """ Clear all feature flags Args: user: Optional user for user-specific flags """ if user: self._clear_all_user_flags(user) else: self._clear_all_global_flags() def get_migration_mode(self) -> str: """ Get current migration mode Returns: str: Migration mode (disabled, testing, gradual, complete) """ return self.is_enabled('migration_mode') or 'disabled' def set_migration_mode(self, mode: str): """ Set migration mode and update related flags Args: mode: Migration mode (disabled, testing, gradual, complete) """ valid_modes = ['disabled', 'testing', 'gradual', 'complete'] if mode not in valid_modes: raise ValueError(f"Invalid migration mode: {mode}. Must be one of {valid_modes}") self.set_flag('migration_mode', mode) # Update related flags based on mode if mode == 'disabled': self.set_flag('use_granular_permissions', False) self.set_flag('fallback_to_legacy', True) self.set_flag('enable_role_templates', False) self.set_flag('enable_page_permissions', False) elif mode == 'testing': self.set_flag('use_granular_permissions', True) self.set_flag('fallback_to_legacy', True) self.set_flag('enable_role_templates', True) self.set_flag('enable_page_permissions', True) self.set_flag('debug_permissions', True) elif mode == 'gradual': self.set_flag('use_granular_permissions', True) self.set_flag('fallback_to_legacy', True) self.set_flag('enable_role_templates', True) self.set_flag('enable_page_permissions', True) self.set_flag('enable_permission_dependencies', True) self.set_flag('enable_permission_expiry', True) elif mode == 'complete': self.set_flag('use_granular_permissions', True) self.set_flag('fallback_to_legacy', False) self.set_flag('enable_role_templates', True) self.set_flag('enable_page_permissions', True) self.set_flag('enable_permission_dependencies', True) self.set_flag('enable_permission_expiry', True) self.set_flag('enable_critical_permissions', True) self.set_flag('enable_bulk_permission_updates', True) self.set_flag('enable_permission_analytics', True) def enable_granular_permissions_for_user(self, user, expires_in: Optional[int] = None): """ Enable granular permissions for a specific user (for testing) Args: user: User to enable granular permissions for expires_in: Optional expiration time in seconds """ self.set_flag('use_granular_permissions', True, user, expires_in) self.set_flag('enable_role_templates', True, user, expires_in) self.set_flag('enable_page_permissions', True, user, expires_in) self.set_flag('debug_permissions', True, user, expires_in) def disable_granular_permissions_for_user(self, user): """ Disable granular permissions for a specific user Args: user: User to disable granular permissions for """ self.clear_flag('use_granular_permissions', user) self.clear_flag('enable_role_templates', user) self.clear_flag('enable_page_permissions', user) self.clear_flag('debug_permissions', user) def get_flag_history(self, flag_name: str, user=None) -> List[Dict[str, Any]]: """ Get history of flag changes Args: flag_name: Name of the feature flag user: Optional user for user-specific flags Returns: List of flag change history """ # This would typically be stored in database for persistence # For now, return empty list as this is a basic implementation return [] def _get_user_flag(self, flag_name: str, user) -> Optional[bool]: """Get user-specific flag value""" cache_key = f"{self.cache_prefix}user_{user.id}_{flag_name}" return cache.get(cache_key) def _set_user_flag(self, flag_name: str, user, value: bool, expires_in: Optional[int] = None): """Set user-specific flag value""" cache_key = f"{self.cache_prefix}user_{user.id}_{flag_name}" timeout = expires_in or self.CACHE_TIMEOUT cache.set(cache_key, value, timeout) # Log flag change logger.info(f"Set user flag {flag_name}={value} for user {user.username}") def _clear_user_flag(self, flag_name: str, user): """Clear user-specific flag""" cache_key = f"{self.cache_prefix}user_{user.id}_{flag_name}" cache.delete(cache_key) def _get_all_user_flags(self, user) -> Dict[str, bool]: """Get all user-specific flags""" # This is a simplified implementation # In practice, you'd want to store these in database for persistence flags = {} for flag_name in self.DEFAULT_FLAGS.keys(): value = self._get_user_flag(flag_name, user) if value is not None: flags[flag_name] = value return flags def _clear_all_user_flags(self, user): """Clear all user-specific flags""" for flag_name in self.DEFAULT_FLAGS.keys(): self._clear_user_flag(flag_name, user) def _get_global_flag(self, flag_name: str) -> Optional[bool]: """Get global flag value""" # Check Django settings first settings_value = getattr(settings, f'PERMISSION_FLAG_{flag_name.upper()}', None) if settings_value is not None: return settings_value # Check cache cache_key = f"{self.cache_prefix}global_{flag_name}" return cache.get(cache_key) def _set_global_flag(self, flag_name: str, value: bool, expires_in: Optional[int] = None): """Set global flag value""" cache_key = f"{self.cache_prefix}global_{flag_name}" timeout = expires_in or self.CACHE_TIMEOUT cache.set(cache_key, value, timeout) # Log flag change logger.info(f"Set global flag {flag_name}={value}") def _clear_global_flag(self, flag_name: str): """Clear global flag""" cache_key = f"{self.cache_prefix}global_{flag_name}" cache.delete(cache_key) def _get_all_global_flags(self) -> Dict[str, bool]: """Get all global flags""" flags = {} # Check Django settings for flag_name in self.DEFAULT_FLAGS.keys(): settings_value = getattr(settings, f'PERMISSION_FLAG_{flag_name.upper()}', None) if settings_value is not None: flags[flag_name] = settings_value # Check cache (this is simplified - in practice you'd want better tracking) for flag_name in self.DEFAULT_FLAGS.keys(): if flag_name not in flags: cache_key = f"{self.cache_prefix}global_{flag_name}" value = cache.get(cache_key) if value is not None: flags[flag_name] = value return flags def _clear_all_global_flags(self): """Clear all global flags""" for flag_name in self.DEFAULT_FLAGS.keys(): self._clear_global_flag(flag_name) # Global instance feature_flags = PermissionFeatureFlags() # Convenience functions def is_granular_permissions_enabled(user=None) -> bool: """Check if granular permissions are enabled""" return feature_flags.is_enabled('use_granular_permissions', user) def is_legacy_fallback_enabled(user=None) -> bool: """Check if legacy fallback is enabled""" return feature_flags.is_enabled('fallback_to_legacy', user) def is_permission_caching_enabled(user=None) -> bool: """Check if permission caching is enabled""" return feature_flags.is_enabled('enable_permission_caching', user) def is_debug_permissions_enabled(user=None) -> bool: """Check if permission debugging is enabled""" return feature_flags.is_enabled('debug_permissions', user) def get_migration_mode() -> str: """Get current migration mode""" return feature_flags.get_migration_mode() def set_migration_mode(mode: str): """Set migration mode""" feature_flags.set_migration_mode(mode) def enable_granular_for_user(user, expires_in: Optional[int] = None): """Enable granular permissions for a user""" feature_flags.enable_granular_permissions_for_user(user, expires_in) def disable_granular_for_user(user): """Disable granular permissions for a user""" feature_flags.disable_granular_permissions_for_user(user)