Python脚本如何批量写入配置数据

wen python案例 2026-06-27 246

本文目录导读：

Python脚本如何批量写入配置数据

使用配置文件格式
使用模板引擎批量生成配置
从数据库批量写入
批量写入到环境变量
批量写入到不同的格式
高级批量处理示例
最佳实践建议

我来介绍几种Python批量写入配置数据的方法：

使用配置文件格式

YAML 格式（推荐）

import yaml
import os
# 准备配置数据
configs = [
    {
        'name': 'server1',
        'host': '192.168.1.1',
        'port': 8080,
        'debug': True
    },
    {
        'name': 'server2',
        'host': '192.168.1.2',
        'port': 9090,
        'debug': False
    }
]
# 批量写入YAML文件
def write_yaml_configs(configs, output_dir='configs'):
    os.makedirs(output_dir, exist_ok=True)
    for config in configs:
        filename = f"{output_dir}/{config['name']}.yaml"
        with open(filename, 'w', encoding='utf-8') as f:
            yaml.dump(config, f, default_flow_style=False)
        print(f"已写入: {filename}")
write_yaml_configs(configs)

JSON 格式

import json
def write_json_configs(configs, output_file='configs.json'):
    with open(output_file, 'w', encoding='utf-8') as f:
        json.dump(configs, f, indent=2, ensure_ascii=False)
    print(f"已写入: {output_file}")
# 批量写入JSON
write_json_configs(configs)

使用模板引擎批量生成配置

Jinja2 模板方式

from jinja2 import Template
import os
# 定义配置模板
template_str = """
[server_{{ server_id }}]
host = {{ host }}
port = {{ port }}
debug = {{ debug }}
max_connections = {{ max_connections }}
"""
# 配置数据
servers = [
    {'server_id': 1, 'host': '192.168.1.1', 'port': 8080, 
     'debug': True, 'max_connections': 100},
    {'server_id': 2, 'host': '192.168.1.2', 'port': 9090, 
     'debug': False, 'max_connections': 200},
]
def generate_configs(template_str, data, output_dir='generated_configs'):
    os.makedirs(output_dir, exist_ok=True)
    template = Template(template_str)
    for item in data:
        config_content = template.render(**item)
        filename = f"{output_dir}/server_{item['server_id']}.ini"
        with open(filename, 'w') as f:
            f.write(config_content)
        print(f"已生成: {filename}")
generate_configs(template_str, servers)

从数据库批量写入

import sqlite3
import json
def export_configs_from_db(db_file, output_file='configs.json'):
    conn = sqlite3.connect(db_file)
    cursor = conn.cursor()
    # 创建示例表
    cursor.execute('''
        CREATE TABLE IF NOT EXISTS servers (
            id INTEGER PRIMARY KEY,
            name TEXT,
            host TEXT,
            port INTEGER,
            config TEXT
        )
    ''')
    # 插入示例数据
    sample_data = [
        ('server1', '192.168.1.1', 8080, '{"debug": true}'),
        ('server2', '192.168.1.2', 9090, '{"debug": false}')
    ]
    cursor.executemany(
        'INSERT OR REPLACE INTO servers (name, host, port, config) VALUES (?, ?, ?, ?)',
        sample_data
    )
    # 查询并写入
    cursor.execute('SELECT name, host, port, config FROM servers')
    configs = []
    for row in cursor.fetchall():
        config = {
            'name': row[0],
            'host': row[1],
            'port': row[2],
            'config': json.loads(row[3])
        }
        configs.append(config)
    with open(output_file, 'w') as f:
        json.dump(configs, f, indent=2)
    print(f"已导出 {len(configs)} 个配置到 {output_file}")
    conn.close()
export_configs_from_db('configs.db')

批量写入到环境变量

import os
import json
def batch_set_env_vars(configs, prefix='APP_'):
    """批量设置环境变量"""
    for config in configs:
        for key, value in config.items():
            env_key = f"{prefix}{config['name'].upper()}_{key.upper()}"
            if isinstance(value, (dict, list)):
                os.environ[env_key] = json.dumps(value)
            else:
                os.environ[env_key] = str(value)
            print(f"设置环境变量: {env_key} = {os.environ[env_key]}")
# 使用示例
batch_set_env_vars(configs)

批量写入到不同的格式

import csv
import os
def batch_write_multiformat(configs, base_dir='configs'):
    """同时写入多种格式"""
    os.makedirs(base_dir, exist_ok=True)
    # 1. 写入CSV
    with open(f'{base_dir}/configs.csv', 'w', newline='') as f:
        writer = csv.DictWriter(f, fieldnames=configs[0].keys())
        writer.writeheader()
        writer.writerows(configs)
    # 2. 写入INI格式
    import configparser
    config = configparser.ConfigParser()
    for item in configs:
        config[item['name']] = {
            'host': item['host'],
            'port': str(item['port']),
            'debug': str(item['debug'])
        }
    with open(f'{base_dir}/configs.ini', 'w') as f:
        config.write(f)
    print("已生成所有格式的配置文件")
batch_write_multiformat(configs)

高级批量处理示例

import concurrent.futures
import time
def write_single_config(config, output_dir):
    """写入单个配置文件（用于并行处理）"""
    time.sleep(0.1)  # 模拟IO操作
    filename = f"{output_dir}/{config['name']}.json"
    with open(filename, 'w') as f:
        json.dump(config, f, indent=2)
    return f"已写入: {filename}"
def parallel_batch_write(configs, output_dir='parallel_configs'):
    """并行批量写入"""
    os.makedirs(output_dir, exist_ok=True)
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
        futures = [
            executor.submit(write_single_config, config, output_dir)
            for config in configs
        ]
        for future in concurrent.futures.as_completed(futures):
            print(future.result())
# 并行写入
parallel_batch_write(configs)

最佳实践建议

选择合适格式：
- YAML：适合复杂配置，可读性好
- JSON：适合程序解析
- INI：适合简单配置

使用配置管理库：

# 使用python-dotenv管理环境变量
from dotenv import load_dotenv, set_key
def update_env_file(env_file, configs):
    for config in configs:
        for key, value in config.items():
            set_key(env_file, key.upper(), str(value))
update_env_file('.env', configs)

错误处理：

def safe_batch_write(configs):
    succeeded = []
    failed = []
    for config in configs:
        try:
            write_single_config(config, 'configs')
            succeeded.append(config['name'])
        except Exception as e:
            failed.append((config['name'], str(e)))
    print(f"成功: {len(succeeded)}, 失败: {len(failed)}")
    return succeeded, failed

根据具体需求选择合适的方案，如果需要更详细的示例,请告诉我具体的使用场景。

本文地址： https://www.bmjup.com/post/10344.html