| """ |
| Configuration management for ARF Demo |
| Updated for better Pydantic compatibility and fallback handling |
| """ |
| from typing import Optional, Dict, Any, List |
| from enum import Enum |
| import os |
| import logging |
|
|
| logger = logging.getLogger(__name__) |
|
|
| |
| try: |
| from pydantic_settings import BaseSettings |
| from pydantic import Field, field_validator, ConfigDict |
| PYDANTIC_V2 = True |
| logger.info("Using pydantic-settings for BaseSettings") |
| except ImportError: |
| try: |
| from pydantic import BaseSettings, Field, validator |
| PYDANTIC_V2 = False |
| logger.info("Using pydantic.BaseSettings (older version)") |
| except ImportError as e: |
| logger.warning(f"Failed to import pydantic: {e}. Using fallback settings.") |
| |
| class BaseSettings: |
| model_config = {} |
| def __init__(self, **kwargs): |
| for k, v in kwargs.items(): |
| setattr(self, k, v) |
| |
| class Field: |
| @staticmethod |
| def default(value): |
| return value |
| |
| def validator(*args, **kwargs): |
| def decorator(func): |
| return func |
| return decorator |
| |
| def field_validator(*args, **kwargs): |
| def decorator(func): |
| return func |
| return decorator |
| |
| PYDANTIC_V2 = False |
|
|
|
|
| class ARFMode(str, Enum): |
| """ARF operation modes""" |
| DEMO = "demo" |
| OSS = "oss" |
| ENTERPRISE = "enterprise" |
|
|
|
|
| class SafetyMode(str, Enum): |
| """Safety modes for execution""" |
| ADVISORY = "advisory" |
| APPROVAL = "approval" |
| AUTONOMOUS = "autonomous" |
|
|
|
|
| class Settings(BaseSettings): |
| """ |
| Application settings with environment variable support |
| """ |
| |
| |
| arf_mode: ARFMode = Field( |
| default=ARFMode.DEMO, |
| description="ARF operation mode" |
| ) |
| |
| use_mock_arf: bool = Field( |
| default=True, |
| description="Use mock ARF implementation (for demo mode)" |
| ) |
| |
| |
| arf_api_key: Optional[str] = Field( |
| default=None, |
| description="ARF API key for real integration" |
| ) |
| |
| arf_base_url: str = Field( |
| default="https://api.arf.dev", |
| description="ARF API base URL" |
| ) |
| |
| |
| engineer_hourly_rate: float = Field( |
| default=150.0, |
| description="Engineer hourly rate in USD" |
| ) |
| |
| engineer_annual_cost: float = Field( |
| default=125000.0, |
| description="Engineer annual cost in USD" |
| ) |
| |
| default_savings_rate: float = Field( |
| default=0.82, |
| description="Default savings rate with ARF" |
| ) |
| |
| |
| auto_refresh_seconds: int = Field( |
| default=30, |
| description="Auto-refresh interval in seconds" |
| ) |
| |
| max_history_items: int = Field( |
| default=100, |
| description="Maximum history items to display" |
| ) |
| |
| |
| default_scenario: str = Field( |
| default="Cache Miss Storm", |
| description="Default incident scenario" |
| ) |
| |
| scenario_config_path: str = Field( |
| default="config/scenarios", |
| description="Path to scenario configuration files" |
| ) |
| |
| |
| default_safety_mode: SafetyMode = Field( |
| default=SafetyMode.ADVISORY, |
| description="Default safety mode" |
| ) |
| |
| require_approval: bool = Field( |
| default=True, |
| description="Require human approval for execution" |
| ) |
| |
| |
| if PYDANTIC_V2: |
| @field_validator("arf_api_key") |
| @classmethod |
| def validate_api_key(cls, v: Optional[str], info) -> Optional[str]: |
| if info.data.get("arf_mode") == ARFMode.ENTERPRISE and not v: |
| raise ValueError("ARF API key required for Enterprise mode") |
| return v |
| |
| @field_validator("use_mock_arf") |
| @classmethod |
| def validate_mock_mode(cls, v: bool, info) -> bool: |
| if info.data.get("arf_mode") == ARFMode.DEMO: |
| return True |
| return v |
| else: |
| @validator("arf_api_key") |
| def validate_api_key(cls, v: Optional[str], values: Dict[str, Any]) -> Optional[str]: |
| if values.get("arf_mode") == ARFMode.ENTERPRISE and not v: |
| raise ValueError("ARF API key required for Enterprise mode") |
| return v |
| |
| @validator("use_mock_arf") |
| def validate_mock_mode(cls, v: bool, values: Dict[str, Any]) -> bool: |
| if values.get("arf_mode") == ARFMode.DEMO: |
| return True |
| return v |
| |
| |
| if PYDANTIC_V2: |
| model_config = ConfigDict( |
| env_file=".env", |
| env_file_encoding="utf-8", |
| case_sensitive=False, |
| use_enum_values=True, |
| extra="ignore" |
| ) |
| else: |
| class Config: |
| env_file = ".env" |
| env_file_encoding = "utf-8" |
| case_sensitive = False |
| use_enum_values = True |
|
|
|
|
| |
| try: |
| settings = Settings() |
| logger.info("Settings loaded successfully") |
| except Exception as e: |
| logger.warning(f"Failed to load settings from .env: {e}, using defaults") |
| |
| settings = Settings( |
| arf_mode=ARFMode.DEMO, |
| use_mock_arf=True, |
| arf_api_key=None, |
| arf_base_url="https://api.arf.dev", |
| engineer_hourly_rate=150.0, |
| engineer_annual_cost=125000.0, |
| default_savings_rate=0.82, |
| auto_refresh_seconds=30, |
| max_history_items=100, |
| default_scenario="Cache Miss Storm", |
| scenario_config_path="config/scenarios", |
| default_safety_mode=SafetyMode.ADVISORY, |
| require_approval=True |
| ) |
|
|
|
|
| def get_settings() -> Settings: |
| """Get settings instance (singleton pattern)""" |
| return settings |
|
|
|
|
| def print_settings_summary() -> None: |
| """Print a summary of current settings (for debugging)""" |
| summary = { |
| "mode": settings.arf_mode.value, |
| "mock_mode": settings.use_mock_arf, |
| "default_scenario": settings.default_scenario, |
| "safety_mode": settings.default_safety_mode.value, |
| "requires_approval": settings.require_approval |
| } |
| logger.info(f"Settings summary: {summary}") |