Config validation

app/config/config.py

@@ -0,0 +1,197 @@
+from typing import Dict, Any, Optional
+
+from app.config.environment import get_environment_config, EnvironmentConfig
+from app.config.loader import load_and_validate_config, get_legacy_config_dict, ConfigValidationError
+from app.config.validation import ImpulseConfig
+from app.logging import logger
+
+
+class UnifiedConfig:
+    """
+    Unified configuration that combines environment and validated application config.
+    Uses existing ImpulseConfig as source of truth for application configuration.
+    """
+
+    def __init__(self, env: EnvironmentConfig, app: ImpulseConfig, legacy: Dict[str, Any]):
+        self.env = env
+        self.app = app
+        self.legacy = legacy
+
+        self.INCIDENT_ACTUAL_VERSION = 'v3.0.0'
+        self.check_updates = True
+
+    @property
+    def settings(self) -> Dict[str, Any]:
+        return self.legacy
+
+    @property
+    def incident(self):
+        return self.app.incident
+
+    @property
+    def experimental(self):
+        return self.app.experimental
+
+    @property
+    def application(self):
+        return self.app.application
+
+    @property
+    def ui_config(self):
+        return self.app.ui
+
+    @property
+    def slack_bot_user_oauth_token(self) -> str:
+        return self.env.slack_bot_user_oauth_token
+
+    @property
+    def slack_verification_token(self) -> str:
+        return self.env.slack_verification_token
+
+    @property
+    def mattermost_access_token(self) -> str:
+        return self.env.mattermost_access_token
+
+    @property
+    def telegram_bot_token(self) -> str:
+        return self.env.telegram_bot_token
+
+    @property
+    def data_path(self) -> str:
+        return self.env.data_path
+
+    @property
+    def config_path(self) -> str:
+        return self.env.config_path
+
+    @property
+    def incidents_path(self) -> str:
+        return self.env.incidents_path
+
+    @property
+    def provider_sync_interval(self) -> int:
+        return self.env.provider_sync_interval
+
+    @property
+    def provider_max_events(self) -> int:
+        return self.env.provider_max_events
+
+    @property
+    def provider_days_to_sync(self) -> int:
+        return self.env.provider_days_to_sync
+
+    @property
+    def provider_service_account_file(self) -> str:
+        return self.env.provider_service_account_file
+
+    @property
+    def cors_allowed_origins(self) -> list:
+        return self.env.cors_allowed_origins
+
+
+_config: Optional[UnifiedConfig] = None
+
+
+def get_config() -> UnifiedConfig:
+    """
+    Get the global configuration instance.
+    """
+    global _config
+    if _config is None:
+        _config = load_unified_config()
+    return _config
+
+
+def load_unified_config(config_path: Optional[str] = None, exit_on_error: bool = True) -> UnifiedConfig:
+    """
+    Load and create unified configuration from environment and YAML file.
+
+    Args:
+        config_path: Optional path to configuration file. Uses env CONFIG_PATH if not provided.
+        exit_on_error: If True, exit process on validation errors. If False, raise exception.
+
+    Returns:
+        UnifiedConfig: Complete configuration object
+
+    Raises:
+        ConfigValidationError: If configuration loading or validation fails
+        SystemExit: If configuration is invalid and exit_on_error is True
+    """
+    try:
+        env_config = get_environment_config()
+
+        if config_path is None:
+            config_path = env_config.config_file_path
+
+        validated_config, raw_config = load_and_validate_config(config_path)
+
+        legacy_config = get_legacy_config_dict(validated_config)
+
+        return UnifiedConfig(
+            env=env_config,
+            app=validated_config,
+            legacy=legacy_config
+        )
+
+    except ConfigValidationError as e:
+        error_msg = (f"{e}\n"
+                    f"Please check your impulse.yml file and fix any validation errors.\n"
+                    f"Documentation: https://docs.impulse.bot/latest/config_file/")
+        if exit_on_error:
+            logger.error(error_msg)
+            raise SystemExit(1)
+        else:
+            logger.warning(error_msg)
+            raise
+    except Exception as e:
+        error_msg = f"Failed to load configuration: {e}"
+        if exit_on_error:
+            logger.error(error_msg)
+            raise SystemExit(1)
+        else:
+            logger.warning(error_msg)
+            raise
+
+
+def reload_config(config_path: Optional[str] = None) -> bool:
+    """
+    Reload configuration from file with graceful error handling.
+    If validation fails, keeps the current configuration and logs a warning.
+
+    Args:
+        config_path: Optional path to configuration file. Uses env CONFIG_PATH if not provided.
+
+    Returns:
+        bool: True if reload was successful, False if failed and kept old config
+    """
+    global _config
+    current_config = _config
+
+    try:
+        new_config = load_unified_config(config_path, exit_on_error=False)
+        if new_config.app.application.type == current_config.app.application.type:
+            _config = new_config
+            logger.info("Configuration reloaded successfully")
+            return True
+        else:
+            logger.warning("Application type changed, keeping current configuration")
+            return False
+
+    except ConfigValidationError as e:
+        logger.warning("Configuration validation failed, keeping current configuration")
+        _config = current_config
+        return False
+    except Exception as e:
+        logger.warning(f"Configuration reload failed, keeping current configuration: {e}")
+        _config = current_config
+        return False
+
+
+def force_reload_config(config_path: Optional[str] = None) -> UnifiedConfig:
+    """
+    Force reload configuration from file (original behavior).
+    Useful for testing or when you want the process to exit on validation errors.
+    """
+    global _config
+    _config = load_unified_config(config_path)
+    return _config

app/config/environment.py

@@ -0,0 +1,126 @@
+import os
+from typing import List
+from pydantic import BaseModel, Field, field_validator
+from dotenv import load_dotenv
+
+# Load environment variables from .env file
+load_dotenv()
+
+
+class EnvironmentConfig(BaseModel):
+    """Environment-based configuration loaded from environment variables"""
+
+    # Authentication tokens and secrets
+    slack_bot_user_oauth_token: str = Field(
+        default_factory=lambda: os.getenv('SLACK_BOT_USER_OAUTH_TOKEN', ''),
+        description="Slack Bot User OAuth Token"
+    )
+    slack_verification_token: str = Field(
+        default_factory=lambda: os.getenv('SLACK_VERIFICATION_TOKEN', ''),
+        description="Slack Verification Token"
+    )
+    mattermost_access_token: str = Field(
+        default_factory=lambda: os.getenv('MATTERMOST_ACCESS_TOKEN', ''),
+        description="Mattermost Access Token"
+    )
+    telegram_bot_token: str = Field(
+        default_factory=lambda: os.getenv('TELEGRAM_BOT_TOKEN', ''),
+        description="Telegram Bot Token"
+    )
+
+    # Paths
+    data_path: str = Field(
+        default_factory=lambda: os.getenv('DATA_PATH', './data'),
+        description="Path to data directory"
+    )
+    config_path: str = Field(
+        default_factory=lambda: os.getenv('CONFIG_PATH', './'),
+        description="Path to configuration directory"
+    )
+
+    # Provider settings (for Google Calendar integration)
+    provider_sync_interval: int = Field(
+        default_factory=lambda: int(os.getenv('CHAIN_PROVIDER_SYNC_INTERVAL_SECONDS', '60')),
+        description="Provider sync interval in seconds"
+    )
+    provider_max_events: int = Field(
+        default_factory=lambda: int(os.getenv('CHAIN_PROVIDER_MAX_EVENTS', '10')),
+        description="Maximum events to sync from provider"
+    )
+    provider_days_to_sync: int = Field(
+        default_factory=lambda: int(os.getenv('CHAIN_PROVIDER_DAYS_TO_SYNC', '7')),
+        description="Number of days to sync from provider"
+    )
+    provider_service_account_file: str = Field(
+        default_factory=lambda: os.getenv('GOOGLE_SERVICE_ACCOUNT_FILE', './key.json'),
+        description="Path to Google service account file"
+    )
+
+    # CORS configuration
+    cors_allowed_origins: List[str] = Field(
+        default_factory=lambda: os.getenv('CORS_ALLOWED_ORIGINS', 'http://localhost:5000').split(','),
+        description="Comma-separated list of allowed CORS origins"
+    )
+
+    # Logging
+    log_level: str = Field(
+        default_factory=lambda: os.getenv('LOG_LEVEL', 'INFO'),
+        description="Logging level"
+    )
+
+    @field_validator('provider_sync_interval', 'provider_max_events', 'provider_days_to_sync')
+    @classmethod
+    def validate_positive_integers(cls, v):
+        """Validate that provider settings are positive integers"""
+        if v <= 0:
+            raise ValueError("Provider configuration values must be positive integers")
+        return v
+
+    @field_validator('cors_allowed_origins')
+    @classmethod
+    def validate_cors_origins(cls, v):
+        """Clean up CORS origins by removing whitespace"""
+        return [origin.strip() for origin in v if origin.strip()]
+
+    @field_validator('log_level')
+    @classmethod
+    def validate_log_level(cls, v):
+        """Validate log level is valid"""
+        valid_levels = ['DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL']
+        if v.upper() not in valid_levels:
+            raise ValueError(f"Log level must be one of: {', '.join(valid_levels)}")
+        return v.upper()
+
+    @property
+    def incidents_path(self) -> str:
+        """Computed property for incidents path"""
+        return f"{self.data_path}/incidents"
+
+    @property
+    def config_file_path(self) -> str:
+        """Computed property for config file path"""
+        return f"{self.config_path}/impulse.yml"
+
+
+# Global instance - created once and reused
+_env_config: EnvironmentConfig = None
+
+
+def get_environment_config() -> EnvironmentConfig:
+    """Get the singleton instance of environment configuration"""
+    global _env_config
+    if _env_config is None:
+        _env_config = EnvironmentConfig()
+    return _env_config
+
+
+# Convenience function for common environment variables
+def get_messenger_token(messenger_type: str) -> str:
+    """Get the appropriate token based on messenger type"""
+    env_config = get_environment_config()
+    token_map = {
+        'slack': env_config.slack_bot_user_oauth_token,
+        'mattermost': env_config.mattermost_access_token,
+        'telegram': env_config.telegram_bot_token,
+    }
+    return token_map.get(messenger_type, '')