""" ARF Settings Configuration with Pydantic 2.x Compatibility Fixed for both OSS and Enterprise deployments """ import os import sys from enum import Enum from typing import Optional, Dict, Any, List, Union from datetime import datetime # Try to import pydantic with version detection try: from pydantic import Field, validator, model_validator from pydantic_settings import BaseSettings, SettingsConfigDict PYDANTIC_V2 = True except ImportError: # Fallback for pydantic v1 from pydantic import Field, validator from pydantic_settings import BaseSettings PYDANTIC_V2 = False import logging logger = logging.getLogger(__name__) # Enums for ARF modes class ARFMode(str, Enum): """ARF Operational Modes""" DEMO = "demo" OSS = "oss" ENTERPRISE = "enterprise" HYBRID = "hybrid" SIMULATION = "simulation" class SafetyMode(str, Enum): """Safety and Compliance Modes""" STANDARD = "standard" STRICT = "strict" PERMISSIVE = "permissive" ENTERPRISE = "enterprise" COMPLIANCE = "compliance" class InstallationStatus(str, Enum): """Package Installation Status""" NOT_INSTALLED = "not_installed" OSS_ONLY = "oss_only" ENTERPRISE_ONLY = "enterprise_only" FULL_INSTALL = "full_install" DEMO_MODE = "demo_mode" # Pydantic 2.x compatible Settings class if PYDANTIC_V2: class Settings(BaseSettings): """ARF Settings with Pydantic 2.x compatibility""" model_config = SettingsConfigDict( env_file=".env", env_file_encoding="utf-8", env_prefix="ARF_", case_sensitive=False, extra="ignore" ) # Core Settings arf_mode: ARFMode = Field(default=ARFMode.DEMO, description="ARF operational mode") safety_mode: SafetyMode = Field(default=SafetyMode.STANDARD, description="Safety compliance mode") debug: bool = Field(default=False, description="Enable debug logging") log_level: str = Field(default="INFO", description="Logging level") # Version Info arf_version: str = Field(default="3.3.7", description="ARF version") demo_version: str = Field(default="3.8.0", description="Demo app version") # Installation Status oss_installed: bool = Field(default=False, description="ARF OSS package installed") enterprise_installed: bool = Field(default=False, description="Enterprise package installed") installation_status: InstallationStatus = Field( default=InstallationStatus.DEMO_MODE, description="Overall installation status" ) # Feature Flags enable_telemetry: bool = Field(default=True, description="Enable telemetry collection") enable_anomaly_detection: bool = Field(default=True, description="Enable anomaly detection") enable_business_metrics: bool = Field(default=True, description="Enable business metrics") enable_audit_trail: bool = Field(default=True, description="Enable audit trail") # Paths data_dir: str = Field(default="./data", description="Data directory") log_dir: str = Field(default="./logs", description="Log directory") cache_dir: str = Field(default="./cache", description="Cache directory") scenario_config_path: str = Field(default="./config/scenarios", description="Path to scenario configuration files") # ADDED THIS # API Settings api_timeout: int = Field(default=30, description="API timeout in seconds") max_retries: int = Field(default=3, description="Maximum API retries") rate_limit: int = Field(default=100, description="API rate limit per minute") # UI Settings ui_theme: str = Field(default="dark", description="UI theme") refresh_interval: int = Field(default=5, description="UI refresh interval in seconds") # Business Settings default_currency: str = Field(default="USD", description="Default currency") cost_per_incident: float = Field(default=10000.0, description="Default cost per incident") engineer_hourly_rate: float = Field(default=150.0, description="Engineer hourly rate") @model_validator(mode='before') @classmethod def validate_enums(cls, data: Any) -> Any: """Validate and convert enum fields from strings""" if isinstance(data, dict): # Convert string values to enums if needed if 'arf_mode' in data and isinstance(data['arf_mode'], str): try: data['arf_mode'] = ARFMode(data['arf_mode'].lower()) except ValueError: data['arf_mode'] = ARFMode.DEMO if 'safety_mode' in data and isinstance(data['safety_mode'], str): try: data['safety_mode'] = SafetyMode(data['safety_mode'].lower()) except ValueError: data['safety_mode'] = SafetyMode.STANDARD if 'installation_status' in data and isinstance(data['installation_status'], str): try: data['installation_status'] = InstallationStatus(data['installation_status'].lower()) except ValueError: data['installation_status'] = InstallationStatus.DEMO_MODE return data @model_validator(mode='after') def check_installation_status(self): """Update installation status based on detected packages""" # Check if OSS is installed try: import agentic_reliability_framework self.oss_installed = True except ImportError: self.oss_installed = False # Check if Enterprise is installed try: import agentic_reliability_enterprise self.enterprise_installed = True except ImportError: self.enterprise_installed = False # Update installation status if self.oss_installed and self.enterprise_installed: self.installation_status = InstallationStatus.FULL_INSTALL elif self.oss_installed: self.installation_status = InstallationStatus.OSS_ONLY elif self.enterprise_installed: self.installation_status = InstallationStatus.ENTERPRISE_ONLY else: self.installation_status = InstallationStatus.DEMO_MODE return self else: # Pydantic v1 compatible Settings class class Settings(BaseSettings): """ARF Settings with Pydantic v1 compatibility""" class Config: env_file = ".env" env_file_encoding = "utf-8" env_prefix = "ARF_" case_sensitive = False # Core Settings arf_mode: ARFMode = Field(default=ARFMode.DEMO, description="ARF operational mode") safety_mode: SafetyMode = Field(default=SafetyMode.STANDARD, description="Safety compliance mode") debug: bool = Field(default=False, description="Enable debug logging") log_level: str = Field(default="INFO", description="Logging level") # Version Info arf_version: str = Field(default="3.3.7", description="ARF version") demo_version: str = Field(default="3.8.0", description="Demo app version") # Installation Status oss_installed: bool = Field(default=False, description="ARF OSS package installed") enterprise_installed: bool = Field(default=False, description="Enterprise package installed") installation_status: InstallationStatus = Field( default=InstallationStatus.DEMO_MODE, description="Overall installation status" ) # Feature Flags enable_telemetry: bool = Field(default=True, description="Enable telemetry collection") enable_anomaly_detection: bool = Field(default=True, description="Enable anomaly detection") enable_business_metrics: bool = Field(default=True, description="Enable business metrics") enable_audit_trail: bool = Field(default=True, description="Enable audit trail") # Paths data_dir: str = Field(default="./data", description="Data directory") log_dir: str = Field(default="./logs", description="Log directory") cache_dir: str = Field(default="./cache", description="Cache directory") scenario_config_path: str = Field(default="./config/scenarios", description="Path to scenario configuration files") # ADDED THIS # API Settings api_timeout: int = Field(default=30, description="API timeout in seconds") max_retries: int = Field(default=3, description="Maximum API retries") rate_limit: int = Field(default=100, description="API rate limit per minute") # UI Settings ui_theme: str = Field(default="dark", description="UI theme") refresh_interval: int = Field(default=5, description="UI refresh interval in seconds") # Business Settings default_currency: str = Field(default="USD", description="Default currency") cost_per_incident: float = Field(default=10000.0, description="Default cost per incident") engineer_hourly_rate: float = Field(default=150.0, description="Engineer hourly rate") @validator('arf_mode', 'safety_mode', 'installation_status', pre=True) def validate_enum_strings(cls, v, field): """Convert string values to enums""" if isinstance(v, str): try: if field.name == 'arf_mode': return ARFMode(v.lower()) elif field.name == 'safety_mode': return SafetyMode(v.lower()) elif field.name == 'installation_status': return InstallationStatus(v.lower()) except ValueError: # Return default value if field.name == 'arf_mode': return ARFMode.DEMO elif field.name == 'safety_mode': return SafetyMode.STANDARD elif field.name == 'installation_status': return InstallationStatus.DEMO_MODE return v @validator('installation_status', always=True) def update_installation_status(cls, v, values): """Update installation status based on detected packages""" # Check if OSS is installed try: import agentic_reliability_framework values['oss_installed'] = True except ImportError: values['oss_installed'] = False # Check if Enterprise is installed try: import agentic_reliability_enterprise values['enterprise_installed'] = True except ImportError: values['enterprise_installed'] = False # Determine installation status if values.get('oss_installed') and values.get('enterprise_installed'): return InstallationStatus.FULL_INSTALL elif values.get('oss_installed'): return InstallationStatus.OSS_ONLY elif values.get('enterprise_installed'): return InstallationStatus.ENTERPRISE_ONLY else: return InstallationStatus.DEMO_MODE # Singleton settings instance try: settings = Settings() logger.info("✅ Settings initialized successfully") logger.info(f"📊 ARF Mode: {settings.arf_mode}") logger.info(f"🛡️ Safety Mode: {settings.safety_mode}") logger.info(f"📦 Installation Status: {settings.installation_status}") except Exception as e: logger.error(f"❌ Failed to initialize settings: {e}") # Create fallback settings class FallbackSettings: arf_mode = ARFMode.DEMO safety_mode = SafetyMode.STANDARD debug = False log_level = "INFO" arf_version = "3.3.7" demo_version = "3.8.0" oss_installed = False enterprise_installed = False installation_status = InstallationStatus.DEMO_MODE enable_telemetry = True enable_anomaly_detection = True enable_business_metrics = True enable_audit_trail = True data_dir = "./data" log_dir = "./logs" cache_dir = "./cache" scenario_config_path = "./config/scenarios" # ADDED THIS api_timeout = 30 max_retries = 3 rate_limit = 100 ui_theme = "dark" refresh_interval = 5 default_currency = "USD" cost_per_incident = 10000.0 engineer_hourly_rate = 150.0 settings = FallbackSettings() logger.warning("⚠️ Using fallback settings due to initialization error") def get_settings() -> Settings: """Get the settings singleton instance""" return settings # Export everything __all__ = [ "Settings", "get_settings", "ARFMode", "SafetyMode", "InstallationStatus", "settings" ]