Python Dynamic Configuration Manager with Runtime Reload

Last update on December 21 2024 07:17:58 (UTC/GMT +8 hours)

Write a Python program to create a dynamic configuration manager that reloads settings at runtime.

The problem involves creating a dynamic configuration manager that can reload settings from a configuration file at runtime without restart. This system should monitor the configuration file for any changes and automatically update the in-memory settings when modifications are detected. It ensures that applications can adapt to new configurations immediately, enhancing flexibility and reducing downtime. Implementing this involves using file system monitoring tools, thread-safe operations, and handling runtime updates seamlessly.

Sample Solution:

config_manager.py

Python Code :

# Import necessary modules
import json
import threading
import time
import signal
import os
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler

# Define the ConfigManager class to manage configurations
class ConfigManager:
    def __init__(self, config_file):
        # Initialize with the configuration file path
        self.config_file = config_file
        # Initialize the configuration data dictionary
        self.config_data = {}
        # Create a lock for thread-safe operations
        self.lock = threading.Lock()
        # Load the initial configuration
        self.load_config()

    def load_config(self):
        # Load the configuration file in a thread-safe manner
        with self.lock:
            with open(self.config_file, 'r') as f:
                self.config_data = json.load(f)
        # Print the loaded configuration for debugging
        print("Configuration reloaded:", self.config_data)

    def get_config(self):
        # Get the current configuration in a thread-safe manner
        with self.lock:
            return self.config_data

    def start_watching(self):
        # Create a file system event handler for the configuration file
        event_handler = ConfigFileHandler(self, self.config_file)
        # Create an observer to monitor file system changes
        observer = Observer()
        # Schedule the observer to watch the current directory for changes
        observer.schedule(event_handler, path='.', recursive=False)
        # Start the observer
        observer.start()
        print("Started watching for configuration changes...")
        
        try:
            # Continuously check if the stop event is set
            while not stop_event.is_set():
                time.sleep(1)
        except KeyboardInterrupt:
            pass
        finally:
            # Stop the observer when shutting down
            print("Stopping observer...")
            observer.stop()
            observer.join()
            print("Observer stopped.")

# Define the file system event handler class for configuration file changes
class ConfigFileHandler(FileSystemEventHandler):
    def __init__(self, config_manager, config_file):
        # Initialize with the config manager and the configuration file path
        self.config_manager = config_manager
        self.config_file = os.path.abspath(config_file)

    def on_modified(self, event):
        # Check if the modified file is the configuration file
        if os.path.abspath(event.src_path) == self.config_file:
            # Print the modification event for debugging
            print(f"Configuration file {event.src_path} changed, reloading...")
            # Reload the configuration
            self.config_manager.load_config()

# Define a signal handler for SIGINT to handle graceful shutdown
def handle_sigint(signal, frame):
    global stop_event
    print("Shutting down...")
    # Set the stop event to terminate the program
    stop_event.set()

# Example usage
if __name__ == "__main__":
    # Define the path to the configuration file
    config_file = 'config.json'
    # Create an instance of ConfigManager with the configuration file
    config_manager = ConfigManager(config_file)

    # Create an event to signal stopping
    stop_event = threading.Event()
    # Register the signal handler for SIGINT
    signal.signal(signal.SIGINT, handle_sigint)
    
    # Start watching for configuration changes in a separate thread
    watcher_thread = threading.Thread(target=config_manager.start_watching)
    watcher_thread.start()

    # Simulate application usage
    try:
        while not stop_event.is_set():
            # Get and print the current configuration periodically
            config = config_manager.get_config()
            print("Current configuration:", config)
            time.sleep(5)
    except KeyboardInterrupt:
        pass

    # Signal the watcher thread to stop
    stop_event.set()
    # Wait for the watcher thread to finish
    watcher_thread.join()
    print("Program terminated.")

config.json

Python Code :

{
    "setting100": "value100",
    "setting200": "value200"
}

Output:

(base) C:\Users\ME>python config_manager.py
Configuration reloaded: {'setting1': 'value100', 'setting2': 'value200'}
Current configuration: {'setting1': 'value100', 'setting2': 'value200'}
Started watching for configuration changes...
Current configuration: {'setting1': 'value100', 'setting2': 'value200'}
Current configuration: {'setting1': 'value100', 'setting2': 'value200'}
Current configuration: {'setting1': 'value100', 'setting2': 'value200'}
Current configuration: {'setting1': 'value100', 'setting2': 'value200'}
Configuration file .\config.json changed, reloading...
Configuration reloaded: {'setting1': 'value1', 'setting2': 'value2'}
Current configuration: {'setting1': 'value1', 'setting2': 'value2'}
Current configuration: {'setting1': 'value1', 'setting2': 'value2'}
Current configuration: {'setting1': 'value1', 'setting2': 'value2'}
Configuration file .\config.json changed, reloading...
Configuration reloaded: {'setting100': 'value1', 'setting200': 'value2'}
Current configuration: {'setting100': 'value1', 'setting200': 'value2'}
Current configuration: {'setting100': 'value1', 'setting200': 'value2'}
Shutting down...
Stopping observer...
Observer stopped.
Program terminated.

(base) C:\Users\ME>

Explanation:

• Imports: Necessary modules are imported, including JSON for parsing, threading for concurrent execution, signal for handling interrupts, and watchdog for monitoring file changes.
• ConfigManager class: Manages the configuration file, loading and providing access to configuration settings in a thread-safe manner.
• ConfigFileHandler class: Handles file modification events to reload the configuration file when it changes.
• handle_sigint function: Gracefully handles shutdown when a SIGINT (Ctrl+C) is detected.
• Main execution block: Initializes the configuration manager, sets up a signal handler for graceful shutdown, starts a thread to watch for file changes, and periodically prints the current configuration until interrupted.

Python Code Editor :

Have another way to solve this solution? Contribute your code (and comments) through Disqus.

Previous: Python Mathematical expression Parsing and Evaluation Library.
Next: Python Genetic Algorithm for Optimization.