Update config_loader.py

config_loader.py

@@ -1,230 +1,233 @@
-# SPDX-License-Identifier: GPL-3.0-or-later
-#
-# Toolify: Empower any LLM with function calling capabilities.
-# Copyright (C) 2025 FunnyCups (https://github.com/funnycups)
-
-import os
-import yaml
-from typing import List, Dict, Any, Set, Optional
-from pydantic import BaseModel, Field, field_validator
-
-
-class ServerConfig(BaseModel):
-    """Server configuration"""
-    port: int = Field(default=8000, ge=1, le=65535, description="Server port")
-    host: str = Field(default="0.0.0.0", description="Server host address")
-    timeout: int = Field(default=180, ge=1, description="Request timeout (seconds)")
-
-
-class UpstreamService(BaseModel):
-    """Upstream service configuration"""
-    name: str = Field(description="Service name")
-    base_url: str = Field(description="Service base URL")
-    api_key: str = Field(description="API key")
-    models: List[str] = Field(description="List of supported models")
-    description: str = Field(default="", description="Service description")
-    is_default: bool = Field(default=False, description="Is default service")
-    
-    @field_validator('base_url')
-    def validate_base_url(cls, v):
-        if not v.startswith(('http://', 'https://')):
-            raise ValueError('base_url must start with http:// or https://')
-        return v.rstrip('/')
-    
-    @field_validator('api_key')
-    def validate_api_key(cls, v):
-        if not v or v.strip() == "":
-            raise ValueError('api_key cannot be empty')
-        return v
-    
-    @field_validator('models')
-    def validate_models(cls, v):
-        if not v or len(v) == 0:
-            raise ValueError('models list cannot be empty')
-        for model in v:
-            if not model or model.strip() == "":
-                raise ValueError('model name cannot be empty')
-        return v
-
-
-class ClientAuthConfig(BaseModel):
-    """Client authentication configuration"""
-    allowed_keys: List[str] = Field(description="List of allowed client API keys")
-    
-    @field_validator('allowed_keys')
-    def validate_allowed_keys(cls, v):
-        if not v or len(v) == 0:
-            raise ValueError('allowed_keys cannot be empty')
-        for key in v:
-            if not key or key.strip() == "":
-                raise ValueError('API key cannot be empty')
-        return v
-
-
-class FeaturesConfig(BaseModel):
-    """Feature configuration"""
-    enable_function_calling: bool = Field(default=True, description="Enable function calling")
-    log_level: str = Field(default="INFO", description="Logging level: DEBUG, INFO, WARNING, ERROR, CRITICAL, or DISABLED")
-    convert_developer_to_system: bool = Field(default=True, description="Convert developer role to system role")
-    prompt_template: Optional[str] = Field(default=None, description="Custom prompt template for function calling")
-    key_passthrough: bool = Field(default=False, description="If true, directly forward client-provided API key to upstream instead of using configured upstream key")
-    model_passthrough: bool = Field(default=False, description="If true, forward all requests directly to the 'openai' upstream service, ignoring model-based routing")
-
-    @field_validator('log_level')
-    def validate_log_level(cls, v):
-        valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "DISABLED"]
-        if v.upper() not in valid_levels:
-            raise ValueError(f"log_level must be one of {valid_levels}")
-        return v.upper()
-
-    @field_validator('prompt_template')
-    def validate_prompt_template(cls, v):
-        if v:
-            if "{tools_list}" not in v or "{trigger_signal}" not in v:
-                raise ValueError("prompt_template must contain {tools_list} and {trigger_signal} placeholders")
-        return v
-
-
-class AppConfig(BaseModel):
-    """Application full configuration"""
-    server: ServerConfig = Field(default_factory=ServerConfig)
-    upstream_services: List[UpstreamService] = Field(description="List of upstream services")
-    client_authentication: ClientAuthConfig = Field(description="Client authentication configuration")
-    features: FeaturesConfig = Field(default_factory=FeaturesConfig)
-    
-    @field_validator('upstream_services')
-    def validate_upstream_services(cls, v):
-        if not v or len(v) == 0:
-            raise ValueError('upstream_services cannot be empty')
-        
-        default_services = [service for service in v if service.is_default]
-        if len(default_services) == 0:
-            raise ValueError('Must have at least one default upstream service (is_default: true)')
-        if len(default_services) > 1:
-            raise ValueError('Only one upstream service can be marked as default')
-        
-        all_models = set()
-        all_aliases = set()
-        
-        for service in v:
-            for model in service.models:
-                if model in all_models:
-                    raise ValueError(f'Duplicate model entry found: {model}')
-                all_models.add(model)
-                
-                if ':' in model:
-                    parts = model.split(':', 1)
-                    if len(parts) == 2:
-                        alias, real_model = parts
-                        if not alias.strip() or not real_model.strip():
-                            raise ValueError(f"Invalid alias format in '{model}'. Both parts must not be empty.")
-                        all_aliases.add(alias)
-                    else:
-                        raise ValueError(f"Invalid model format with colon: {model}")
-
-        regular_models = {m for m in all_models if ':' not in m}
-        conflicts = all_aliases.intersection(regular_models)
-        if conflicts:
-            raise ValueError(f"Alias names {conflicts} conflict with model names.")
-                
-        return v
-
-
-class ConfigLoader:
-    """Configuration loader"""
-    
-    def __init__(self, config_path: str = "config.yaml"):
-        self.config_path = config_path
-        self._config: AppConfig = None
-    
-    def load_config(self) -> AppConfig:
-        """Load configuration file"""
-        if not os.path.exists(self.config_path):
-            raise FileNotFoundError(
-                f"Configuration file '{self.config_path}' not found. "
-                f"Please copy 'config.example.yaml' to '{self.config_path}' and modify the configuration as needed."
-            )
-        
-        try:
-            with open(self.config_path, 'r', encoding='utf-8') as f:
-                config_data = yaml.safe_load(f)
-        except yaml.YAMLError as e:
-            raise ValueError(f"Configuration file format error: {e}")
-        except Exception as e:
-            raise ValueError(f"Failed to read configuration file: {e}")
-        
-        if not config_data:
-            raise ValueError("Configuration file is empty")
-        
-        try:
-            self._config = AppConfig(**config_data)
-            return self._config
-        except Exception as e:
-            raise ValueError(f"Configuration validation failed: {e}")
-    
-    @property
-    def config(self) -> AppConfig:
-        """Get configuration object"""
-        if self._config is None:
-            self.load_config()
-        return self._config
-    
-    def get_model_to_service_mapping(self) -> tuple[Dict[str, Dict[str, Any]], Dict[str, List[str]]]:
-        """Get model to service mapping and model aliases"""
-        config = self.config
-        model_mapping = {}
-        alias_mapping = {}
-        
-        for service in config.upstream_services:
-            service_info = {
-                "name": service.name,
-                "base_url": service.base_url,
-                "api_key": service.api_key,
-                "description": service.description,
-                "is_default": service.is_default
-            }
-            
-            for model_entry in service.models:
-                model_mapping[model_entry] = service_info
-                if ':' in model_entry:
-                    parts = model_entry.split(':', 1)
-                    if len(parts) == 2:
-                        alias, _ = parts
-                        if alias not in alias_mapping:
-                            alias_mapping[alias] = []
-                        alias_mapping[alias].append(model_entry)
-        
-        return model_mapping, alias_mapping
-    
-    def get_default_service(self) -> Dict[str, Any]:
-        """Get default service configuration"""
-        config = self.config
-        for service in config.upstream_services:
-            if service.is_default:
-                return {
-                    "name": service.name,
-                    "base_url": service.base_url,
-                    "api_key": service.api_key,
-                    "description": service.description,
-                    "is_default": service.is_default
-                }
-        raise ValueError("No default service configured")
-    
-    def get_allowed_client_keys(self) -> Set[str]:
-        """Get set of allowed client keys"""
-        return set(self.config.client_authentication.allowed_keys)
-    
-    def get_log_level(self) -> str:
-        """Get configured log level"""
-        return self.config.features.log_level
-    
-    def get_features_config(self) -> Dict[str, Any]:
-        """Get feature configuration"""
-        return {
-            "function_calling": self.config.features.enable_function_calling,
-            "log_level": self.config.features.log_level,
-            "convert_developer_to_system": self.config.features.convert_developer_to_system
-        }
-
-
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# Toolify: Empower any LLM with function calling capabilities.
+# Copyright (C) 2025 FunnyCups (https://github.com/funnycups)
+
+import os
+import yaml
+from typing import List, Dict, Any, Set, Optional
+from pydantic import BaseModel, Field, field_validator
+
+
+class ServerConfig(BaseModel):
+    """Server configuration"""
+    port: int = Field(default=8000, ge=1, le=65535, description="Server port")
+    host: str = Field(default="0.0.0.0", description="Server host address")
+    timeout: int = Field(default=180, ge=1, description="Request timeout (seconds)")
+
+
+class UpstreamService(BaseModel):
+    """Upstream service configuration"""
+    name: str = Field(description="Service name")
+    base_url: str = Field(description="Service base URL")
+    api_key: str = Field(description="API key")
+    models: List[str] = Field(description="List of supported models")
+    description: str = Field(default="", description="Service description")
+    is_default: bool = Field(default=False, description="Is default service")
+    
+    @field_validator('base_url')
+    def validate_base_url(cls, v):
+        if not v.startswith(('http://', 'https://')):
+            raise ValueError('base_url must start with http:// or https://')
+        return v.rstrip('/')
+    
+    @field_validator('api_key')
+    def validate_api_key(cls, v):
+        if not v or v.strip() == "":
+            raise ValueError('api_key cannot be empty')
+        return v
+    
+    @field_validator('models')
+    def validate_models(cls, v):
+        if not v or len(v) == 0:
+            raise ValueError('models list cannot be empty')
+        for model in v:
+            if not model or model.strip() == "":
+                raise ValueError('model name cannot be empty')
+        return v
+
+
+class ClientAuthConfig(BaseModel):
+    """Client authentication configuration"""
+    allowed_keys: List[str] = Field(description="List of allowed client API keys")
+    
+    @field_validator('allowed_keys')
+    def validate_allowed_keys(cls, v):
+        if not v or len(v) == 0:
+            raise ValueError('allowed_keys cannot be empty')
+        for key in v:
+            if not key or key.strip() == "":
+                raise ValueError('API key cannot be empty')
+        return v
+
+
+class FeaturesConfig(BaseModel):
+    """Feature configuration"""
+    enable_function_calling: bool = Field(default=True, description="Enable function calling")
+    log_level: str = Field(default="INFO", description="Logging level: DEBUG, INFO, WARNING, ERROR, CRITICAL, or DISABLED")
+    convert_developer_to_system: bool = Field(default=True, description="Convert developer role to system role")
+    prompt_template: Optional[str] = Field(default=None, description="Custom prompt template for function calling")
+    key_passthrough: bool = Field(default=False, description="If true, directly forward client-provided API key to upstream instead of using configured upstream key")
+    model_passthrough: bool = Field(default=False, description="If true, forward all requests directly to the 'openai' upstream service, ignoring model-based routing")
+
+    @field_validator('log_level')
+    def validate_log_level(cls, v):
+        valid_levels = ["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL", "DISABLED"]
+        if v.upper() not in valid_levels:
+            raise ValueError(f"log_level must be one of {valid_levels}")
+        return v.upper()
+
+    @field_validator('prompt_template')
+    def validate_prompt_template(cls, v):
+        if v:
+            if "{tools_list}" not in v or "{trigger_signal}" not in v:
+                raise ValueError("prompt_template must contain {tools_list} and {trigger_signal} placeholders")
+        return v
+
+
+class AppConfig(BaseModel):
+    """Application full configuration"""
+    server: ServerConfig = Field(default_factory=ServerConfig)
+    upstream_services: List[UpstreamService] = Field(description="List of upstream services")
+    client_authentication: ClientAuthConfig = Field(description="Client authentication configuration")
+    features: FeaturesConfig = Field(default_factory=FeaturesConfig)
+    
+    @field_validator('upstream_services')
+    def validate_upstream_services(cls, v):
+        if not v or len(v) == 0:
+            raise ValueError('upstream_services cannot be empty')
+        
+        default_services = [service for service in v if service.is_default]
+        if len(default_services) == 0:
+            raise ValueError('Must have at least one default upstream service (is_default: true)')
+        if len(default_services) > 1:
+            raise ValueError('Only one upstream service can be marked as default')
+        
+        all_models = set()
+        all_aliases = set()
+        
+        for service in v:
+            for model in service.models:
+                if model in all_models:
+                    raise ValueError(f'Duplicate model entry found: {model}')
+                all_models.add(model)
+                
+                if ':' in model:
+                    parts = model.split(':', 1)
+                    if len(parts) == 2:
+                        alias, real_model = parts
+                        if not alias.strip() or not real_model.strip():
+                            raise ValueError(f"Invalid alias format in '{model}'. Both parts must not be empty.")
+                        all_aliases.add(alias)
+                    else:
+                        raise ValueError(f"Invalid model format with colon: {model}")
+
+        regular_models = {m for m in all_models if ':' not in m}
+        conflicts = all_aliases.intersection(regular_models)
+        if conflicts:
+            raise ValueError(f"Alias names {conflicts} conflict with model names.")
+                
+        return v
+
+
+class ConfigLoader:
+    """Configuration loader"""
+    
+    def __init__(self, config_path: str = "config.yaml"):
+        env_config_path = os.getenv("CONFIG_PATH")
+        if env_config_path:
+            config_path = env_config_path
+        self.config_path = config_path
+        self._config: AppConfig = None
+    
+    def load_config(self) -> AppConfig:
+        """Load configuration file"""
+        if not os.path.exists(self.config_path):
+            raise FileNotFoundError(
+                f"Configuration file '{self.config_path}' not found. "
+                f"Please copy 'config.example.yaml' to '{self.config_path}' and modify the configuration as needed."
+            )
+        
+        try:
+            with open(self.config_path, 'r', encoding='utf-8') as f:
+                config_data = yaml.safe_load(f)
+        except yaml.YAMLError as e:
+            raise ValueError(f"Configuration file format error: {e}")
+        except Exception as e:
+            raise ValueError(f"Failed to read configuration file: {e}")
+        
+        if not config_data:
+            raise ValueError("Configuration file is empty")
+        
+        try:
+            self._config = AppConfig(**config_data)
+            return self._config
+        except Exception as e:
+            raise ValueError(f"Configuration validation failed: {e}")
+    
+    @property
+    def config(self) -> AppConfig:
+        """Get configuration object"""
+        if self._config is None:
+            self.load_config()
+        return self._config
+    
+    def get_model_to_service_mapping(self) -> tuple[Dict[str, Dict[str, Any]], Dict[str, List[str]]]:
+        """Get model to service mapping and model aliases"""
+        config = self.config
+        model_mapping = {}
+        alias_mapping = {}
+        
+        for service in config.upstream_services:
+            service_info = {
+                "name": service.name,
+                "base_url": service.base_url,
+                "api_key": service.api_key,
+                "description": service.description,
+                "is_default": service.is_default
+            }
+            
+            for model_entry in service.models:
+                model_mapping[model_entry] = service_info
+                if ':' in model_entry:
+                    parts = model_entry.split(':', 1)
+                    if len(parts) == 2:
+                        alias, _ = parts
+                        if alias not in alias_mapping:
+                            alias_mapping[alias] = []
+                        alias_mapping[alias].append(model_entry)
+        
+        return model_mapping, alias_mapping
+    
+    def get_default_service(self) -> Dict[str, Any]:
+        """Get default service configuration"""
+        config = self.config
+        for service in config.upstream_services:
+            if service.is_default:
+                return {
+                    "name": service.name,
+                    "base_url": service.base_url,
+                    "api_key": service.api_key,
+                    "description": service.description,
+                    "is_default": service.is_default
+                }
+        raise ValueError("No default service configured")
+    
+    def get_allowed_client_keys(self) -> Set[str]:
+        """Get set of allowed client keys"""
+        return set(self.config.client_authentication.allowed_keys)
+    
+    def get_log_level(self) -> str:
+        """Get configured log level"""
+        return self.config.features.log_level
+    
+    def get_features_config(self) -> Dict[str, Any]:
+        """Get feature configuration"""
+        return {
+            "function_calling": self.config.features.enable_function_calling,
+            "log_level": self.config.features.log_level,
+            "convert_developer_to_system": self.config.features.convert_developer_to_system
+        }
+
+
 config_loader = ConfigLoader()