"""
Configuration system for the OllamaTools plugin architecture.
This module provides classes and functions for managing plugin configuration.
"""
import os
import sys
import yaml
import json
import logging
from typing import Dict, Any, List, Optional, Set, Union, Callable
from ollamatools.plugins.errors import PluginConfigurationError
# Configure logging
logger = logging.getLogger(__name__)
class PluginConfig:
"""Configuration manager for plugins."""
def __init__(self, config=None, schema=None):
"""
Initialize the configuration manager.
Args:
config: Initial configuration dictionary or path to config file
schema: Optional schema for validation
"""
self.schema = schema or {}
self.config = {}
if config:
self.load_config(config)
def load_config(self, config):
"""
Load configuration from a dictionary, JSON, or YAML file.
Args:
config: Configuration source (dict, path to JSON/YAML file)
Returns:
PluginConfig: Self for method chaining
Raises:
PluginConfigurationError: If configuration loading fails
"""
try:
if isinstance(config, dict):
# Load from dictionary
self.config = config
elif isinstance(config, str) and os.path.exists(config):
# Load from file
with open(config, 'r') as f:
if config.endswith('.json'):
self.config = json.load(f)
elif config.endswith(('.yaml', '.yml')):
self.config = yaml.safe_load(f)
else:
raise PluginConfigurationError(
f"Unsupported configuration file format: {config}",
details={"file": config}
)
else:
raise PluginConfigurationError(
f"Invalid configuration source: {config}",
details={"source": str(config)}
)
# Validate the configuration
if self.schema:
self.validate()
return self
except PluginConfigurationError:
# Re-raise plugin configuration errors
raise
except Exception as e:
# Wrap other exceptions
raise PluginConfigurationError(
f"Failed to load configuration: {str(e)}",
details={"source": str(config)}
) from e
def validate(self):
"""
Validate the configuration against the schema.
Returns:
bool: True if valid
Raises:
PluginConfigurationError: If validation fails
"""
# Simple schema validation
for key, schema_item in self.schema.items():
# Check required fields
if schema_item.get('required', False) and key not in self.config:
raise PluginConfigurationError(
f"Missing required configuration key: {key}",
details={"key": key, "schema": schema_item}
)
# Check type if the key exists
if key in self.config:
value = self.config[key]
expected_type = schema_item.get('type')
if expected_type and not isinstance(value, expected_type):
raise PluginConfigurationError(
f"Invalid type for configuration key {key}. Expected {expected_type.__name__}, got {type(value).__name__}",
details={"key": key, "expected_type": expected_type.__name__, "actual_type": type(value).__name__}
)
# Check enum values
enum_values = schema_item.get('enum')
if enum_values and value not in enum_values:
raise PluginConfigurationError(
f"Invalid value for configuration key {key}. Expected one of {enum_values}, got {value}",
details={"key": key, "expected_values": enum_values, "actual_value": value}
)
# Check custom validation
validator = schema_item.get('validator')
if validator and callable(validator):
if not validator(value):
raise PluginConfigurationError(
f"Validation failed for configuration key {key}",
details={"key": key, "value": value}
)
return True
def get(self, key, default=None):
"""
Get a configuration value.
Args:
key: Configuration key
default: Default value if key is not found
Returns:
Any: Configuration value or default
"""
return self.config.get(key, default)
def set(self, key, value):
"""
Set a configuration value.
Args:
key: Configuration key
value: Configuration value
Returns:
PluginConfig: Self for method chaining
Raises:
PluginConfigurationError: If validation fails
"""
self.config[key] = value
# Validate if schema exists for this key
if self.schema and key in self.schema:
schema_item = self.schema[key]
# Check type
expected_type = schema_item.get('type')
if expected_type and not isinstance(value, expected_type):
raise PluginConfigurationError(
f"Invalid type for configuration key {key}. Expected {expected_type.__name__}, got {type(value).__name__}",
details={"key": key, "expected_type": expected_type.__name__, "actual_type": type(value).__name__}
)
# Check enum values
enum_values = schema_item.get('enum')
if enum_values and value not in enum_values:
raise PluginConfigurationError(
f"Invalid value for configuration key {key}. Expected one of {enum_values}, got {value}",
details={"key": key, "expected_values": enum_values, "actual_value": value}
)
# Check custom validation
validator = schema_item.get('validator')
if validator and callable(validator):
if not validator(value):
raise PluginConfigurationError(
f"Validation failed for configuration key {key}",
details={"key": key, "value": value}
)
return self
def merge(self, other_config):
"""
Merge another configuration into this one.
Args:
other_config: Another PluginConfig or dictionary
Returns:
PluginConfig: Self for method chaining
"""
if isinstance(other_config, PluginConfig):
other_dict = other_config.to_dict()
else:
other_dict = other_config
# Deep merge
self._deep_merge(self.config, other_dict)
# Validate the merged configuration
if self.schema:
self.validate()
return self
def _deep_merge(self, target, source):
"""
Deep merge two dictionaries.
Args:
target: Target dictionary
source: Source dictionary
"""
for key, value in source.items():
if key in target and isinstance(target[key], dict) and isinstance(value, dict):
self._deep_merge(target[key], value)
else:
target[key] = value
def to_dict(self):
"""
Convert the configuration to a dictionary.
Returns:
Dict[str, Any]: Configuration dictionary
"""
return self.config.copy()
def save(self, file_path, format='yaml'):
"""
Save the configuration to a file.
Args:
file_path: Path to save the configuration to
format: File format ('json' or 'yaml')
Returns:
PluginConfig: Self for method chaining
Raises:
PluginConfigurationError: If saving fails
"""
try:
# Create directory if it doesn't exist
os.makedirs(os.path.dirname(file_path), exist_ok=True)
with open(file_path, 'w') as f:
if format.lower() == 'json':
json.dump(self.config, f, indent=2)
elif format.lower() in ('yaml', 'yml'):
yaml.dump(self.config, f, default_flow_style=False)
else:
raise PluginConfigurationError(
f"Unsupported configuration format: {format}",
details={"format": format}
)
return self
except PluginConfigurationError:
# Re-raise plugin configuration errors
raise
except Exception as e:
# Wrap other exceptions
raise PluginConfigurationError(
f"Failed to save configuration: {str(e)}",
details={"file_path": file_path, "format": format}
) from e
def create_default_config(plugin_name, config_type='shared'):
"""
Create a default configuration for a plugin.
Args:
plugin_name: Name of the plugin
config_type: Type of configuration ('shared', 'matrix', 'agent')
Returns:
Dict[str, Any]: Default configuration
"""
# Base configuration
config = {
"enabled": True,
"name": plugin_name
}
# Add type-specific defaults
if config_type == 'shared':
pass # No additional defaults
elif config_type == 'matrix':
config.update({
"commands_enabled": True,
"command_prefix": "/"
})
elif config_type == 'agent':
config.update({
"functions_enabled": True
})
return config
def load_plugin_config(plugin_name, config_dir=None):
"""
Load configuration for a plugin from the default location.
Args:
plugin_name: Name of the plugin
config_dir: Optional custom configuration directory
Returns:
PluginConfig: Plugin configuration
"""
if not config_dir:
# Use default location
config_dir = os.path.join(
os.path.expanduser('~'),
'.ollamatools',
'plugins'
)
# Check for plugin-specific config file
config_file = os.path.join(config_dir, f"{plugin_name.lower()}.yaml")
if os.path.exists(config_file):
return PluginConfig(config_file)
else:
# Return default configuration
return PluginConfig(create_default_config(plugin_name))