Python脚本如何动态修改配置参数

本文目录导读：

1. 使用配置文件（JSON/YAML）
2. 使用环境变量
3. 使用数据库存储配置
4. 热重载配置（使用watchdog监听文件变化）
5. 使用configparser（INI格式）
6. 命令行参数动态修改
7. 完整的动态配置系统示例
8. 选择建议

Python动态修改配置参数有多种方法,以下是几种常用的实现方式：

使用配置文件（JSON/YAML）

JSON配置文件

import json
class ConfigManager:
    def __init__(self, config_path='config.json'):
        self.config_path = config_path
        self.config = self.load_config()
    def load_config(self):
        with open(self.config_path, 'r') as f:
            return json.load(f)
    def get(self, key, default=None):
        return self.config.get(key, default)
    def set(self, key, value):
        self.config[key] = value
        self.save_config()
    def save_config(self):
        with open(self.config_path, 'w') as f:
            json.dump(self.config, f, indent=4)
# 使用示例
config = ConfigManager()
print(config.get('database.host'))
config.set('debug', True)

YAML配置文件

import yaml
class YamlConfig:
    def __init__(self, config_path='config.yaml'):
        self.config_path = config_path
        self.config = self.load_config()
    def load_config(self):
        with open(self.config_path, 'r') as f:
            return yaml.safe_load(f)
    def update(self, key, value):
        # 支持嵌套键，如 'database.host'
        keys = key.split('.')
        current = self.config
        for k in keys[:-1]:
            current = current.setdefault(k, {})
        current[keys[-1]] = value
        self.save_config()
    def save_config(self):
        with open(self.config_path, 'w') as f:
            yaml.dump(self.config, f, default_flow_style=False)

使用环境变量

import os
from dotenv import load_dotenv, set_key
class EnvConfig:
    def __init__(self, env_file='.env'):
        self.env_file = env_file
        load_dotenv(env_file)
    def get(self, key, default=None):
        return os.getenv(key, default)
    def set(self, key, value):
        # 运行时修改
        os.environ[key] = str(value)
        # 持久化到.env文件
        set_key(self.env_file, key, str(value))

使用数据库存储配置

import sqlite3
class DBConfig:
    def __init__(self, db_path='config.db'):
        self.conn = sqlite3.connect(db_path)
        self.create_table()
    def create_table(self):
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS config (
                key TEXT PRIMARY KEY,
                value TEXT
            )
        ''')
        self.conn.commit()
    def get(self, key, default=None):
        cursor = self.conn.execute(
            'SELECT value FROM config WHERE key = ?', (key,)
        )
        result = cursor.fetchone()
        return result[0] if result else default
    def set(self, key, value):
        self.conn.execute(
            'INSERT OR REPLACE INTO config (key, value) VALUES (?, ?)',
            (key, str(value))
        )
        self.conn.commit()

热重载配置（使用watchdog监听文件变化）

import time
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
class ConfigWatcher(FileSystemEventHandler):
    def __init__(self, config_path, callback):
        self.config_path = config_path
        self.callback = callback
    def on_modified(self, event):
        if event.src_path == self.config_path:
            print(f"配置文件已修改: {event.src_path}")
            self.callback()
class DynamicConfig:
    def __init__(self, config_path='config.json'):
        self.config_path = config_path
        self.config = self.load_config()
        self.start_watcher()
    def load_config(self):
        with open(self.config_path, 'r') as f:
            return json.load(f)
    def reload_config(self):
        self.config = self.load_config()
        print("配置已重新加载")
    def start_watcher(self):
        event_handler = ConfigWatcher(
            self.config_path, 
            self.reload_config
        )
        observer = Observer()
        observer.schedule(event_handler, path='.', recursive=False)
        observer.start()

使用configparser（INI格式）

import configparser
class INIConfig:
    def __init__(self, config_path='config.ini'):
        self.config_path = config_path
        self.config = configparser.ConfigParser()
        self.config.read(config_path)
    def get(self, section, key):
        return self.config.get(section, key)
    def set(self, section, key, value):
        if not self.config.has_section(section):
            self.config.add_section(section)
        self.config.set(section, key, value)
        self.save()
    def save(self):
        with open(self.config_path, 'w') as f:
            self.config.write(f)

命令行参数动态修改

import argparse
class CommandLineConfig:
    def __init__(self):
        self.parser = argparse.ArgumentParser()
        self.args = None
    def add_argument(self, *args, **kwargs):
        self.parser.add_argument(*args, **kwargs)
    def parse(self):
        self.args = self.parser.parse_args()
        return self.args
    def update_from_command(self, config_dict):
        # 从命令行参数更新配置
        for key, value in vars(self.args).items():
            if value is not None:
                config_dict[key] = value

完整的动态配置系统示例

import json
import threading
import time
class DynamicConfigSystem:
    def __init__(self, config_path='config.json', auto_reload=True):
        self.config_path = config_path
        self.config = {}
        self.lock = threading.Lock()
        self.observers = []
        self.load_config()
        if auto_reload:
            self.start_auto_reload()
    def load_config(self):
        try:
            with open(self.config_path, 'r') as f:
                with self.lock:
                    self.config = json.load(f)
        except FileNotFoundError:
            self.config = {}
    def get(self, key, default=None):
        with self.lock:
            keys = key.split('.')
            value = self.config
            for k in keys:
                if isinstance(value, dict):
                    value = value.get(k)
                else:
                    return default
            return value if value is not None else default
    def set(self, key, value):
        with self.lock:
            keys = key.split('.')
            current = self.config
            for k in keys[:-1]:
                if k not in current:
                    current[k] = {}
                current = current[k]
            current[keys[-1]] = value
            self.save_config()
            self.notify_observers(key, value)
    def save_config(self):
        with open(self.config_path, 'w') as f:
            json.dump(self.config, f, indent=4)
    def add_observer(self, callback):
        self.observers.append(callback)
    def notify_observers(self, key, value):
        for observer in self.observers:
            observer(key, value)
    def start_auto_reload(self, interval=5):
        def reload_loop():
            last_mtime = None
            while True:
                try:
                    current_mtime = os.path.getmtime(self.config_path)
                    if last_mtime is not None and current_mtime > last_mtime:
                        self.load_config()
                        self.notify_observers('*', None)
                    last_mtime = current_mtime
                except Exception:
                    pass
                time.sleep(interval)
        thread = threading.Thread(target=reload_loop, daemon=True)
        thread.start()
# 使用示例
config = DynamicConfigSystem()
def on_config_change(key, value):
    print(f"配置变更: {key} = {value}")
config.add_observer(on_config_change)
print(config.get('database.host', 'localhost'))
config.set('debug_mode', True)

选择建议

• 简单应用：使用JSON/YAML配置文件
• 容器化部署：环境变量方式
• 分布式系统：数据库或配置中心（如Consul, etcd）
• 需要热更新：文件监控+重载机制
• 安全性要求高：环境变量+加密文件

根据具体场景选择合适的动态配置方案，关键是要保证线程安全、持久化和实时性。