"""
Pattern 278: Configuration Management
Purpose: Dynamic configuration with hot-reload, validation, and versioning
"""
from typing import Dict, Any, List, Optional, Callable, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import copy
class ConfigSource(Enum):
"""Source of configuration."""
FILE = "file"
ENVIRONMENT = "environment"
DATABASE = "database"
API = "api"
DEFAULT = "default"
class ConfigScope(Enum):
"""Scope of configuration."""
GLOBAL = "global"
ENVIRONMENT_SPECIFIC = "environment_specific" # dev, staging, prod
USER_SPECIFIC = "user_specific"
SESSION_SPECIFIC = "session_specific"
@dataclass
class ConfigValue:
"""Represents a configuration value."""
key: str
value: Any
source: ConfigSource
scope: ConfigScope
version: int = 1
last_modified: datetime = field(default_factory=datetime.now)
previous_value: Optional[Any] = None
metadata: Dict[str, Any] = field(default_factory=dict)
@dataclass
class ConfigChange:
"""Record of configuration change."""
key: str
old_value: Any
new_value: Any
changed_by: str
timestamp: datetime = field(default_factory=datetime.now)
reason: Optional[str] = None
class ConfigurationManagement:
"""
Implements configuration management pattern.
Features:
- Dynamic configuration updates
- Hot-reload without restart
- Configuration validation
- Version tracking
- Change history
- Environment-specific configs
- Rollback capability
- Change notifications
"""
def __init__(self):
self.configs: Dict[str, ConfigValue] = {}
self.change_history: List[ConfigChange] = []
self.validators: Dict[str, Callable] = {}
self.change_listeners: List[Callable] = []
self.environments = {"dev", "staging", "prod"}
self.current_environment = "dev"
# Statistics
self.statistics = {
"total_configs": 0,
"total_changes": 0,
"hot_reloads": 0,
"validation_failures": 0,
"rollbacks": 0
}
def set_config(
self,
key: str,
value: Any,
source: ConfigSource = ConfigSource.DEFAULT,
scope: ConfigScope = ConfigScope.GLOBAL,
changed_by: str = "system",
reason: Optional[str] = None
) -> Dict[str, Any]:
"""
Set a configuration value.
Args:
key: Configuration key
value: Configuration value
source: Source of configuration
scope: Scope of configuration
changed_by: Who made the change
reason: Reason for change
Returns:
Status of configuration set
"""
# Validate if validator exists
if key in self.validators:
validator = self.validators[key]
try:
is_valid = validator(value)
if not is_valid:
self.statistics["validation_failures"] += 1
return {
"success": False,
"message": f"Validation failed for key '{key}'"
}
except Exception as e:
self.statistics["validation_failures"] += 1
return {
"success": False,
"message": f"Validation error: {str(e)}"
}
# Record change if existing
if key in self.configs:
old_config = self.configs[key]
change = ConfigChange(
key=key,
old_value=old_config.value,
new_value=value,
changed_by=changed_by,
reason=reason
)
self.change_history.append(change)
self.statistics["total_changes"] += 1
# Update config
old_config.previous_value = old_config.value
old_config.value = value
old_config.version += 1
old_config.last_modified = datetime.now()
old_config.source = source
# Notify listeners
self._notify_listeners(key, old_config.value, value)
return {
"success": True,
"key": key,
"old_value": old_config.previous_value,
"new_value": value,
"version": old_config.version,
"message": f"Configuration '{key}' updated"
}
else:
# Create new config
config = ConfigValue(
key=key,
value=value,
source=source,
scope=scope
)
self.configs[key] = config
self.statistics["total_configs"] += 1
# Notify listeners
self._notify_listeners(key, None, value)
return {
"success": True,
"key": key,
"value": value,
"version": 1,
"message": f"Configuration '{key}' created"
}
def get_config(
self,
key: str,
default: Optional[Any] = None,
environment: Optional[str] = None
) -> Any:
"""
Get a configuration value.
Args:
key: Configuration key
default: Default value if not found
environment: Specific environment to get config for
Returns:
Configuration value or default
"""
env = environment or self.current_environment
# Try environment-specific key first
env_key = f"{key}.{env}"
if env_key in self.configs:
return self.configs[env_key].value
# Fall back to global key
if key in self.configs:
return self.configs[key].value
return default
def hot_reload(self, configs: Dict[str, Any], changed_by: str = "system") -> Dict[str, Any]:
"""
Hot-reload multiple configurations without restart.
Args:
configs: Dictionary of key-value pairs to update
changed_by: Who initiated the reload
Returns:
Status of hot-reload
"""
success_count = 0
failed_count = 0
failures = []
for key, value in configs.items():
result = self.set_config(
key=key,
value=value,
source=ConfigSource.API,
changed_by=changed_by,
reason="hot_reload"
)
if result["success"]:
success_count += 1
else:
failed_count += 1
failures.append({"key": key, "error": result["message"]})
self.statistics["hot_reloads"] += 1
return {
"success": failed_count == 0,
"configs_updated": success_count,
"configs_failed": failed_count,
"failures": failures,
"message": f"Hot-reload completed: {success_count} updated, {failed_count} failed"
}
def register_validator(self, key: str, validator: Callable[[Any], bool]) -> Dict[str, Any]:
"""
Register a validator for a configuration key.
Args:
key: Configuration key
validator: Function that returns True if valid
Returns:
Status of validator registration
"""
self.validators[key] = validator
return {
"success": True,
"key": key,
"message": f"Validator registered for '{key}'"
}
def add_change_listener(self, listener: Callable[[str, Any, Any], None]) -> Dict[str, Any]:
"""
Add a listener for configuration changes.
Args:
listener: Function called when config changes (key, old_value, new_value)
Returns:
Status of listener addition
"""
self.change_listeners.append(listener)
return {
"success": True,
"total_listeners": len(self.change_listeners),
"message": "Change listener added"
}
def _notify_listeners(self, key: str, old_value: Any, new_value: Any) -> None:
"""Notify all listeners of configuration change."""
for listener in self.change_listeners:
try:
listener(key, old_value, new_value)
except Exception as e:
# Log but don't fail
pass
def rollback_config(self, key: str) -> Dict[str, Any]:
"""
Rollback a configuration to its previous value.
Args:
key: Configuration key to rollback
Returns:
Status of rollback
"""
if key not in self.configs:
return {
"success": False,
"message": f"Configuration '{key}' not found"
}
config = self.configs[key]
if config.previous_value is None:
return {
"success": False,
"message": f"No previous value available for '{key}'"
}
old_value = config.value
config.value = config.previous_value
config.previous_value = None
config.version += 1
config.last_modified = datetime.now()
# Record rollback
change = ConfigChange(
key=key,
old_value=old_value,
new_value=config.value,
changed_by="system",
reason="rollback"
)
self.change_history.append(change)
self.statistics["rollbacks"] += 1
# Notify listeners
self._notify_listeners(key, old_value, config.value)
return {
"success": True,
"key": key,
"rolled_back_to": config.value,
"message": f"Configuration '{key}' rolled back"
}
def get_config_info(self, key: str) -> Dict[str, Any]:
"""Get detailed information about a configuration."""
if key not in self.configs:
return {
"success": False,
"message": f"Configuration '{key}' not found"
}
config = self.configs[key]
# Find related changes
changes = [
{
"old_value": c.old_value,
"new_value": c.new_value,
"changed_by": c.changed_by,
"timestamp": c.timestamp.isoformat(),
"reason": c.reason
}
for c in self.change_history if c.key == key
]
return {
"success": True,
"key": key,
"value": config.value,
"source": config.source.value,
"scope": config.scope.value,
"version": config.version,
"last_modified": config.last_modified.isoformat(),
"previous_value": config.previous_value,
"has_validator": key in self.validators,
"change_count": len(changes),
"recent_changes": changes[-5:] # Last 5 changes
}
def list_configs(
self,
scope: Optional[ConfigScope] = None,
source: Optional[ConfigSource] = None
) -> Dict[str, Any]:
"""List all configurations with optional filters."""
configs_list = []
for key, config in self.configs.items():
if scope and config.scope != scope:
continue
if source and config.source != source:
continue
configs_list.append({
"key": key,
"value": config.value,
"source": config.source.value,
"scope": config.scope.value,
"version": config.version
})
return {
"total": len(configs_list),
"configs": configs_list
}
def export_configs(self) -> Dict[str, Any]:
"""Export all configurations."""
return {key: config.value for key, config in self.configs.items()}
def import_configs(
self,
configs: Dict[str, Any],
source: ConfigSource = ConfigSource.FILE
) -> Dict[str, Any]:
"""Import configurations from dictionary."""
imported = 0
for key, value in configs.items():
result = self.set_config(
key=key,
value=value,
source=source,
changed_by="import"
)
if result["success"]:
imported += 1
return {
"success": True,
"imported": imported,
"total": len(configs),
"message": f"Imported {imported}/{len(configs)} configurations"
}
def get_change_history(self, key: Optional[str] = None, limit: int = 10) -> Dict[str, Any]:
"""Get configuration change history."""
if key:
changes = [c for c in self.change_history if c.key == key]
else:
changes = self.change_history
changes_list = [
{
"key": c.key,
"old_value": c.old_value,
"new_value": c.new_value,
"changed_by": c.changed_by,
"timestamp": c.timestamp.isoformat(),
"reason": c.reason
}
for c in changes[-limit:]
]
return {
"total": len(changes),
"showing": len(changes_list),
"changes": changes_list
}
def get_statistics(self) -> Dict[str, Any]:
"""Get configuration management statistics."""
return {
"total_configs": self.statistics["total_configs"],
"total_changes": self.statistics["total_changes"],
"hot_reloads": self.statistics["hot_reloads"],
"validation_failures": self.statistics["validation_failures"],
"rollbacks": self.statistics["rollbacks"],
"registered_validators": len(self.validators),
"change_listeners": len(self.change_listeners)
}
class ConfigurationManagementPattern:
"""
Coordinator for configuration management.
Manages dynamic configuration with hot-reload and validation.
"""
def __init__(self):
self.management = ConfigurationManagement()
def set_config(self, key: str, value: Any, **kwargs) -> Dict[str, Any]:
"""Set configuration."""
return self.management.set_config(key, value, **kwargs)
def get_config(self, key: str, default: Optional[Any] = None) -> Any:
"""Get configuration."""
return self.management.get_config(key, default)
def hot_reload(self, configs: Dict[str, Any], changed_by: str = "system") -> Dict[str, Any]:
"""Hot-reload configurations."""
return self.management.hot_reload(configs, changed_by)
def register_validator(self, key: str, validator: Callable) -> Dict[str, Any]:
"""Register validator."""
return self.management.register_validator(key, validator)
def rollback_config(self, key: str) -> Dict[str, Any]:
"""Rollback configuration."""
return self.management.rollback_config(key)
def get_config_info(self, key: str) -> Dict[str, Any]:
"""Get configuration info."""
return self.management.get_config_info(key)
def list_configs(self) -> Dict[str, Any]:
"""List configurations."""
return self.management.list_configs()
def get_statistics(self) -> Dict[str, Any]:
"""Get statistics."""
return self.management.get_statistics()
# Example usage
if __name__ == "__main__":
print("=== Configuration Management Demo ===\n")
# Create configuration management
pattern = ConfigurationManagementPattern()
print("1. Setting initial configurations...")
pattern.set_config("max_workers", 10, source=ConfigSource.DEFAULT)
pattern.set_config("timeout_seconds", 30, source=ConfigSource.DEFAULT)
pattern.set_config("log_level", "INFO", source=ConfigSource.DEFAULT)
pattern.set_config("enable_caching", True, source=ConfigSource.DEFAULT)
configs = pattern.list_configs()
print(f" Total configs: {configs['total']}")
for config in configs['configs']:
print(f" - {config['key']}: {config['value']}")
# Register validator
print("\n2. Registering validator for max_workers...")
pattern.register_validator("max_workers", lambda v: isinstance(v, int) and 1 <= v <= 100)
print(" Validator registered")
# Try valid update
print("\n3. Updating max_workers to 20...")
result = pattern.set_config("max_workers", 20, changed_by="admin", reason="increase capacity")
print(f" {result['message']}")
print(f" Old value: {result['old_value']}, New value: {result['new_value']}")
# Try invalid update
print("\n4. Trying invalid update (max_workers = 200)...")
result = pattern.set_config("max_workers", 200)
print(f" Result: {result['message']}")
# Hot-reload multiple configs
print("\n5. Hot-reloading multiple configurations...")
new_configs = {
"timeout_seconds": 60,
"log_level": "DEBUG",
"enable_caching": False
}
result = pattern.hot_reload(new_configs, changed_by="deployment")
print(f" {result['message']}")
# Check config values
print("\n6. Checking current values...")
print(f" max_workers: {pattern.get_config('max_workers')}")
print(f" timeout_seconds: {pattern.get_config('timeout_seconds')}")
print(f" log_level: {pattern.get_config('log_level')}")
# Get config info
print("\n7. Detailed info for max_workers...")
info = pattern.get_config_info("max_workers")
if info['success']:
print(f" Current value: {info['value']}")
print(f" Version: {info['version']}")
print(f" Has validator: {info['has_validator']}")
print(f" Change count: {info['change_count']}")
# Rollback
print("\n8. Rolling back max_workers...")
result = pattern.rollback_config("max_workers")
if result['success']:
print(f" {result['message']}")
print(f" Value after rollback: {pattern.get_config('max_workers')}")
# Statistics
print("\n9. Configuration statistics:")
stats = pattern.get_statistics()
print(f" Total configs: {stats['total_configs']}")
print(f" Total changes: {stats['total_changes']}")
print(f" Hot reloads: {stats['hot_reloads']}")
print(f" Validation failures: {stats['validation_failures']}")
print(f" Rollbacks: {stats['rollbacks']}")
print("\nā Comprehensive configuration management with hot-reload complete!")
Agentic Architecture Pattern: Configuration Management
Pattern 278: 278 Configuration Management – Line-by-Line Explanation
Overview
Pattern 278: Configuration Management
Purpose: Dynamic configuration with hot-reload, validation, and versioning
Code Structure Analysis
File Information
- Pattern Number: 278
- Pattern Name: 278 Configuration Management
- File Path:
patterns\deployment\278_configuration_management.py - Total Lines: 598
Detailed Code Explanation
Module Docstring
"""
Pattern 278: Configuration Management
Purpose: Dynamic configuration with hot-reload, validation, and versioning
"""
Explanation: This module docstring describes the purpose and key features of the 278 Configuration Management pattern.
Imports Section
from typing import Dict, Any, List, Optional, Callable, Set
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
from collections import defaultdict
import copy # First 20 imports shown
Explanation: The module imports necessary dependencies:
– from typing import Dict, Any, List, Optional, Callable, Set: Type hints for better code documentation
– from dataclasses import dataclass, field: Data class decorators for clean data structures
– from datetime import datetime: Date and time utilities
– from enum import Enum: Enumeration support for fixed sets of values
Enumerations
ConfigSource (Line 15)
class ConfigSource(Enum):
FILE
ENVIRONMENT
DATABASE
API
DEFAULT
Purpose: Source of configuration.
Values:
– FILE: Represents file
– ENVIRONMENT: Represents environment
– DATABASE: Represents database
– API: Represents api
– DEFAULT: Represents default
ConfigScope (Line 24)
class ConfigScope(Enum):
GLOBAL
ENVIRONMENT_SPECIFIC
USER_SPECIFIC
SESSION_SPECIFIC
Purpose: Scope of configuration.
Values:
– GLOBAL: Represents global
– ENVIRONMENT_SPECIFIC: Represents environment specific
– USER_SPECIFIC: Represents user specific
– SESSION_SPECIFIC: Represents session specific
Classes
ConfigSource (Line 15)
Description: Source of configuration.
ConfigScope (Line 24)
Description: Scope of configuration.
ConfigValue (Line 33)
Description: Represents a configuration value.
ConfigChange (Line 46)
Description: Record of configuration change.
ConfigurationManagement (Line 56)
Description: Implements configuration management pattern.
Features: – Dynamic configuration updates – Hot-reload without restart – Configuration validation – Version tracking – Change history – Environment-specific configs – Rollback capability – Change notifications
Methods:
__init__ (Line 71)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
set_config (Line 88)
def set_config(self, key, value, source, scope, changed_by, reason):
"""Set a configuration value.
Args:
key: Configuration key
value: Configuration value
source: Source of configuration
scope: Scope of configuration
changed_by: Who made the change
reason: Reason for change
Returns:
Status of configuration set"""
Purpose: Set a configuration value.
Args: key: Configuration key value: Configuration value source: Source of configuration scope: Scope of configuration changed_by: Who made the change reason: Reason for change
Returns: Status of configuration set
Parameters:
– self: Instance reference
– key: Parameter for key
– value: Parameter for value
– source: Parameter for source
– scope: Parameter for scope
– changed_by: Parameter for changed_by
– reason: Parameter for reason
get_config (Line 182)
def get_config(self, key, default, environment):
"""Get a configuration value.
Args:
key: Configuration key
default: Default value if not found
environment: Specific environment to get config for
Returns:
Configuration value or default"""
Purpose: Get a configuration value.
Args: key: Configuration key default: Default value if not found environment: Specific environment to get config for
Returns: Configuration value or default
Parameters:
– self: Instance reference
– key: Parameter for key
– default: Parameter for default
– environment: Parameter for environment
hot_reload (Line 212)
def hot_reload(self, configs, changed_by):
"""Hot-reload multiple configurations without restart.
Args:
configs: Dictionary of key-value pairs to update
changed_by: Who initiated the reload
Returns:
Status of hot-reload"""
Purpose: Hot-reload multiple configurations without restart.
Args: configs: Dictionary of key-value pairs to update changed_by: Who initiated the reload
Returns: Status of hot-reload
Parameters:
– self: Instance reference
– configs: Parameter for configs
– changed_by: Parameter for changed_by
register_validator (Line 252)
def register_validator(self, key, validator):
"""Register a validator for a configuration key.
Args:
key: Configuration key
validator: Function that returns True if valid
Returns:
Status of validator registration"""
Purpose: Register a validator for a configuration key.
Args: key: Configuration key validator: Function that returns True if valid
Returns: Status of validator registration
Parameters:
– self: Instance reference
– key: Parameter for key
– validator: Parameter for validator
add_change_listener (Line 271)
def add_change_listener(self, listener):
"""Add a listener for configuration changes.
Args:
listener: Function called when config changes (key, old_value, new_value)
Returns:
Status of listener addition"""
Purpose: Add a listener for configuration changes.
Args: listener: Function called when config changes (key, old_value, new_value)
Returns: Status of listener addition
Parameters:
– self: Instance reference
– listener: Parameter for listener
_notify_listeners (Line 289)
def _notify_listeners(self, key, old_value, new_value):
"""Notify all listeners of configuration change."""
Purpose: Notify all listeners of configuration change.
Parameters:
– self: Instance reference
– key: Parameter for key
– old_value: Parameter for old_value
– new_value: Parameter for new_value
rollback_config (Line 298)
def rollback_config(self, key):
"""Rollback a configuration to its previous value.
Args:
key: Configuration key to rollback
Returns:
Status of rollback"""
Purpose: Rollback a configuration to its previous value.
Args: key: Configuration key to rollback
Returns: Status of rollback
Parameters:
– self: Instance reference
– key: Parameter for key
get_config_info (Line 349)
def get_config_info(self, key):
"""Get detailed information about a configuration."""
Purpose: Get detailed information about a configuration.
Parameters:
– self: Instance reference
– key: Parameter for key
list_configs (Line 385)
def list_configs(self, scope, source):
"""List all configurations with optional filters."""
Purpose: List all configurations with optional filters.
Parameters:
– self: Instance reference
– scope: Parameter for scope
– source: Parameter for source
export_configs (Line 412)
def export_configs(self):
"""Export all configurations."""
Purpose: Export all configurations.
Parameters:
– self: Instance reference
import_configs (Line 416)
def import_configs(self, configs, source):
"""Import configurations from dictionary."""
Purpose: Import configurations from dictionary.
Parameters:
– self: Instance reference
– configs: Parameter for configs
– source: Parameter for source
get_change_history (Line 441)
def get_change_history(self, key, limit):
"""Get configuration change history."""
Purpose: Get configuration change history.
Parameters:
– self: Instance reference
– key: Parameter for key
– limit: Parameter for limit
get_statistics (Line 466)
def get_statistics(self):
"""Get configuration management statistics."""
Purpose: Get configuration management statistics.
Parameters:
– self: Instance reference
ConfigurationManagementPattern (Line 479)
Description: Coordinator for configuration management.
Manages dynamic configuration with hot-reload and validation.
Methods:
__init__ (Line 486)
def __init__(self):
"""No method docstring."""
Purpose: No method docstring.
Parameters:
– self: Instance reference
set_config (Line 489)
def set_config(self, key, value):
"""Set configuration."""
Purpose: Set configuration.
Parameters:
– self: Instance reference
– key: Parameter for key
– value: Parameter for value
get_config (Line 493)
def get_config(self, key, default):
"""Get configuration."""
Purpose: Get configuration.
Parameters:
– self: Instance reference
– key: Parameter for key
– default: Parameter for default
hot_reload (Line 497)
def hot_reload(self, configs, changed_by):
"""Hot-reload configurations."""
Purpose: Hot-reload configurations.
Parameters:
– self: Instance reference
– configs: Parameter for configs
– changed_by: Parameter for changed_by
register_validator (Line 501)
def register_validator(self, key, validator):
"""Register validator."""
Purpose: Register validator.
Parameters:
– self: Instance reference
– key: Parameter for key
– validator: Parameter for validator
rollback_config (Line 505)
def rollback_config(self, key):
"""Rollback configuration."""
Purpose: Rollback configuration.
Parameters:
– self: Instance reference
– key: Parameter for key
get_config_info (Line 509)
def get_config_info(self, key):
"""Get configuration info."""
Purpose: Get configuration info.
Parameters:
– self: Instance reference
– key: Parameter for key
list_configs (Line 513)
def list_configs(self):
"""List configurations."""
Purpose: List configurations.
Parameters:
– self: Instance reference
get_statistics (Line 517)
def get_statistics(self):
"""Get statistics."""
Purpose: Get statistics.
Parameters:
– self: Instance reference
Design Patterns Used
- Strategy Pattern: Different execution strategies for various scenarios
- State Pattern: Managing different states in the workflow
- Chain of Responsibility: Sequential processing through workflow nodes
- Observer Pattern: Tracking and monitoring pattern execution
- Factory Pattern: Creating instances with proper configuration
Usage Example
# Initialize the pattern
pattern = ConfigSource(name="example", model_name="gpt-4")
# Configure and execute
result = pattern.execute(data={"key": "value"})
# Check results
print(result)
Key Concepts
- LLM Integration: Uses LangChain/OpenAI for intelligent decision-making
- Workflow Orchestration: LangGraph manages execution flow through nodes
- State Management: TypedDict defines workflow state structure
- Error Handling: Comprehensive error handling and fallback mechanisms
- Extensibility: Designed for easy customization and extension
Extension Points
This pattern can be extended by: 1. Adding custom workflow nodes 2. Implementing additional validation logic 3. Integrating with external services 4. Customizing LLM prompts and responses 5. Adding monitoring and logging capabilities
Leave a Reply