Python脚本如何批量读取配置数据

wen python案例 29

本文目录导读:

Python脚本如何批量读取配置数据

  1. 使用配置文件(多种格式)
  2. 使用环境变量
  3. 使用数据库存储配置
  4. 高级配置管理类
  5. 递归读取目录结构
  6. 实用封装函数
  7. 最佳实践建议

我来介绍几种批量读取配置数据的Python方法:

使用配置文件(多种格式)

JSON 配置文件

import json
import os
def read_json_configs(config_dir='configs'):
    """批量读取JSON配置文件"""
    configs = {}
    for filename in os.listdir(config_dir):
        if filename.endswith('.json'):
            filepath = os.path.join(config_dir, filename)
            with open(filepath, 'r', encoding='utf-8') as f:
                configs[filename] = json.load(f)
    return configs
# 使用示例
configs = read_json_configs()
for name, config in configs.items():
    print(f"{name}: {config}")

YAML 配置文件

import yaml
import os
def read_yaml_configs(config_dir='configs'):
    """批量读取YAML配置文件"""
    configs = {}
    for filename in os.listdir(config_dir):
        if filename.endswith(('.yaml', '.yml')):
            filepath = os.path.join(config_dir, filename)
            with open(filepath, 'r', encoding='utf-8') as f:
                configs[filename] = yaml.safe_load(f)
    return configs

INI 配置文件

import configparser
import os
def read_ini_configs(config_dir='configs'):
    """批量读取INI配置文件"""
    configs = {}
    for filename in os.listdir(config_dir):
        if filename.endswith('.ini'):
            config = configparser.ConfigParser()
            config.read(os.path.join(config_dir, filename))
            configs[filename] = {section: dict(config[section]) 
                                for section in config.sections()}
    return configs

使用环境变量

import os
from dotenv import load_dotenv
def load_env_configs(env_files=None):
    """加载环境变量配置"""
    if env_files:
        for env_file in env_files:
            load_dotenv(env_file)
    else:
        load_dotenv()  # 默认加载 .env
    return {key: os.getenv(key) for key in os.environ}

使用数据库存储配置

SQLite 配置

import sqlite3
import json
class ConfigDB:
    def __init__(self, db_path='configs.db'):
        self.conn = sqlite3.connect(db_path)
        self.create_table()
    def create_table(self):
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS configs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                name TEXT UNIQUE,
                value TEXT,
                type TEXT DEFAULT 'string',
                created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
            )
        ''')
        self.conn.commit()
    def get_all_configs(self):
        cursor = self.conn.execute('SELECT name, value, type FROM configs')
        configs = {}
        for name, value, type_ in cursor:
            if type_ == 'json':
                configs[name] = json.loads(value)
            elif type_ == 'int':
                configs[name] = int(value)
            elif type_ == 'float':
                configs[name] = float(value)
            elif type_ == 'bool':
                configs[name] = value.lower() == 'true'
            else:
                configs[name] = value
        return configs

高级配置管理类

import os
import json
import yaml
import configparser
from typing import Any, Dict
class ConfigManager:
    """统一的配置管理器"""
    SUPPORTED_EXTENSIONS = {
        '.json': 'load_json',
        '.yaml': 'load_yaml', 
        '.yml': 'load_yaml',
        '.ini': 'load_ini',
        '.cfg': 'load_ini'
    }
    def __init__(self, config_dirs=None):
        self.config_dirs = config_dirs or ['configs', './']
        self.configs = {}
        self.load_all()
    def load_json(self, filepath):
        with open(filepath, 'r', encoding='utf-8') as f:
            return json.load(f)
    def load_yaml(self, filepath):
        with open(filepath, 'r', encoding='utf-8') as f:
            return yaml.safe_load(f)
    def load_ini(self, filepath):
        config = configparser.ConfigParser()
        config.read(filepath)
        return {section: dict(config[section]) for section in config.sections()}
    def load_all(self):
        """递归加载所有配置文件"""
        for config_dir in self.config_dirs:
            if not os.path.exists(config_dir):
                continue
            for root, dirs, files in os.walk(config_dir):
                for file in files:
                    ext = os.path.splitext(file)[1].lower()
                    if ext in self.SUPPORTED_EXTENSIONS:
                        filepath = os.path.join(root, file)
                        try:
                            loader = getattr(self, self.SUPPORTED_EXTENSIONS[ext])
                            config_name = os.path.relpath(filepath, config_dir)
                            self.configs[config_name] = loader(filepath)
                            print(f"✓ 加载配置: {config_name}")
                        except Exception as e:
                            print(f"✗ 加载失败: {config_name} - {e}")
    def get(self, key, default=None):
        """获取配置值"""
        return self.configs.get(key, default)
    def get_all(self):
        """获取所有配置"""
        return self.configs
# 使用示例
if __name__ == '__main__':
    # 批量读取配置
    config_manager = ConfigManager(['configs', 'settings'])
    # 获取特定配置
    db_config = config_manager.get('database.json')
    app_config = config_manager.get('app.yaml')
    # 打印所有配置
    for name, config in config_manager.get_all().items():
        print(f"\n📁 {name}:")
        print(json.dumps(config, indent=2, ensure_ascii=False))

递归读取目录结构

import os
from pathlib import Path
def read_config_directory(base_dir='configs'):
    """递归读取配置目录"""
    configs = {}
    base_path = Path(base_dir)
    if not base_path.exists():
        return configs
    for file_path in base_path.rglob('*'):
        if file_path.is_file():
            relative_path = str(file_path.relative_to(base_path))
            # 根据扩展名选择解析方式
            ext = file_path.suffix.lower()
            if ext == '.json':
                with open(file_path, 'r') as f:
                    configs[relative_path] = json.load(f)
            elif ext in ('.yaml', '.yml'):
                with open(file_path, 'r') as f:
                    configs[relative_path] = yaml.safe_load(f)
            # ... 其他格式
    return configs

实用封装函数

def load_all_configs(config_paths=None, recursive=True):
    """便捷的配置加载函数"""
    if config_paths is None:
        config_paths = ['configs', '.env', 'settings']
    all_configs = {}
    for path in config_paths:
        if os.path.isdir(path):
            # 读取目录
            for root, dirs, files in os.walk(path) if recursive else [(path, [], os.listdir(path))]:
                for file in files:
                    filepath = os.path.join(root, file)
                    # 解析文件...
                    pass
        elif os.path.isfile(path):
            # 读取单个文件
            pass
    return all_configs

最佳实践建议

# 推荐的配置组织结构
project/
├── configs/
│   ├── development/
│   │   ├── database.json
│   │   └── app.yaml
│   ├── production/
│   │   ├── database.json
│   │   └── app.yaml
│   └── common/
│       └── logging.json
├── .env
└── config_loader.py

这些方法可以根据你的具体需求选择使用,对于简单的项目,JSON/YAML文件就够了;对于复杂项目,建议使用配置管理类来处理。

抱歉,评论功能暂时关闭!