# app/services/admin_settings_service.py """ Admin settings service for platform-wide configuration. This module provides functions for: - Managing platform settings - Getting/setting configuration values - Encrypting sensitive settings """ import json import logging from datetime import UTC, datetime from typing import Any from sqlalchemy import func from sqlalchemy.orm import Session from app.exceptions import ( AdminOperationException, ResourceNotFoundException, ValidationException, ) from models.database.admin import AdminSetting from models.schema.admin import ( AdminSettingCreate, AdminSettingResponse, AdminSettingUpdate, ) logger = logging.getLogger(__name__) class AdminSettingsService: """Service for managing platform-wide settings.""" def get_setting_by_key(self, db: Session, key: str) -> AdminSetting | None: """Get setting by key.""" try: return ( db.query(AdminSetting) .filter(func.lower(AdminSetting.key) == key.lower()) .first() ) except Exception as e: logger.error(f"Failed to get setting {key}: {str(e)}") return None def get_setting_value(self, db: Session, key: str, default: Any = None) -> Any: """ Get setting value with type conversion. Args: key: Setting key default: Default value if setting doesn't exist Returns: Typed setting value """ setting = self.get_setting_by_key(db, key) if not setting: return default # Convert value based on type try: if setting.value_type == "integer": return int(setting.value) if setting.value_type == "float": return float(setting.value) if setting.value_type == "boolean": return setting.value.lower() in ("true", "1", "yes") if setting.value_type == "json": return json.loads(setting.value) return setting.value except Exception as e: logger.error(f"Failed to convert setting {key} value: {str(e)}") return default def get_all_settings( self, db: Session, category: str | None = None, is_public: bool | None = None, ) -> list[AdminSettingResponse]: """Get all settings with optional filtering.""" try: query = db.query(AdminSetting) if category: query = query.filter(AdminSetting.category == category) if is_public is not None: query = query.filter(AdminSetting.is_public == is_public) settings = query.order_by(AdminSetting.category, AdminSetting.key).all() return [ AdminSettingResponse.model_validate(setting) for setting in settings ] except Exception as e: logger.error(f"Failed to get settings: {str(e)}") raise AdminOperationException( operation="get_all_settings", reason="Database query failed" ) def get_settings_by_category(self, db: Session, category: str) -> dict[str, Any]: """ Get all settings in a category as a dictionary. Returns: Dictionary of key-value pairs """ settings = self.get_all_settings(db, category=category) result = {} for setting in settings: # Convert value based on type if setting.value_type == "integer": result[setting.key] = int(setting.value) elif setting.value_type == "float": result[setting.key] = float(setting.value) elif setting.value_type == "boolean": result[setting.key] = setting.value.lower() in ("true", "1", "yes") elif setting.value_type == "json": result[setting.key] = json.loads(setting.value) else: result[setting.key] = setting.value return result def create_setting( self, db: Session, setting_data: AdminSettingCreate, admin_user_id: int ) -> AdminSettingResponse: """Create new setting.""" try: # Check if setting already exists existing = self.get_setting_by_key(db, setting_data.key) if existing: raise ValidationException( f"Setting with key '{setting_data.key}' already exists" ) # Validate value based on type self._validate_setting_value(setting_data.value, setting_data.value_type) # TODO: Encrypt value if is_encrypted=True value_to_store = setting_data.value if setting_data.is_encrypted: # value_to_store = self._encrypt_value(setting_data.value) pass setting = AdminSetting( key=setting_data.key.lower(), value=value_to_store, value_type=setting_data.value_type, category=setting_data.category, description=setting_data.description, is_encrypted=setting_data.is_encrypted, is_public=setting_data.is_public, last_modified_by_user_id=admin_user_id, ) db.add(setting) db.flush() db.refresh(setting) logger.info(f"Setting '{setting.key}' created by admin {admin_user_id}") return AdminSettingResponse.model_validate(setting) except ValidationException: raise except Exception as e: logger.error(f"Failed to create setting: {str(e)}") raise AdminOperationException( operation="create_setting", reason="Database operation failed" ) def update_setting( self, db: Session, key: str, update_data: AdminSettingUpdate, admin_user_id: int ) -> AdminSettingResponse: """Update existing setting.""" setting = self.get_setting_by_key(db, key) if not setting: raise ResourceNotFoundException(resource_type="setting", identifier=key) try: # Validate new value self._validate_setting_value(update_data.value, setting.value_type) # TODO: Encrypt value if needed value_to_store = update_data.value if setting.is_encrypted: # value_to_store = self._encrypt_value(update_data.value) pass setting.value = value_to_store if update_data.description is not None: setting.description = update_data.description setting.last_modified_by_user_id = admin_user_id setting.updated_at = datetime.now(UTC) db.flush() db.refresh(setting) logger.info(f"Setting '{setting.key}' updated by admin {admin_user_id}") return AdminSettingResponse.model_validate(setting) except ValidationException: raise except Exception as e: logger.error(f"Failed to update setting {key}: {str(e)}") raise AdminOperationException( operation="update_setting", reason="Database operation failed" ) def upsert_setting( self, db: Session, setting_data: AdminSettingCreate, admin_user_id: int ) -> AdminSettingResponse: """Create or update setting (upsert).""" existing = self.get_setting_by_key(db, setting_data.key) if existing: update_data = AdminSettingUpdate( value=setting_data.value, description=setting_data.description ) return self.update_setting(db, setting_data.key, update_data, admin_user_id) return self.create_setting(db, setting_data, admin_user_id) def delete_setting(self, db: Session, key: str, admin_user_id: int) -> str: """Delete setting.""" setting = self.get_setting_by_key(db, key) if not setting: raise ResourceNotFoundException(resource_type="setting", identifier=key) try: db.delete(setting) logger.warning(f"Setting '{key}' deleted by admin {admin_user_id}") return f"Setting '{key}' successfully deleted" except Exception as e: logger.error(f"Failed to delete setting {key}: {str(e)}") raise AdminOperationException( operation="delete_setting", reason="Database operation failed" ) # ============================================================================ # HELPER METHODS # ============================================================================ def _validate_setting_value(self, value: str, value_type: str): """Validate setting value matches declared type.""" try: if value_type == "integer": int(value) elif value_type == "float": float(value) elif value_type == "boolean": if value.lower() not in ("true", "false", "1", "0", "yes", "no"): raise ValueError("Invalid boolean value") elif value_type == "json": json.loads(value) except Exception as e: raise ValidationException( f"Value '{value}' is not valid for type '{value_type}': {str(e)}" ) def _encrypt_value(self, value: str) -> str: """Encrypt sensitive setting value.""" # TODO: Implement encryption using Fernet or similar # from cryptography.fernet import Fernet # return encrypted_value return value def _decrypt_value(self, encrypted_value: str) -> str: """Decrypt sensitive setting value.""" # TODO: Implement decryption return encrypted_value # Create service instance admin_settings_service = AdminSettingsService()