本文目录导读：
在脚本中解析 YAML 配置有多种实现方式,我为你介绍几种主流方法：
Python 解析 YAML
安装 PyYAML
pip install pyyaml
基础解析示例
import yaml
import json
# 读取 YAML 文件
with open('config.yaml', 'r', encoding='utf-8') as file:
config = yaml.safe_load(file)
# 解析示例
print(config['database']['host'])
print(config['server']['port'])
# 将 YAML 转换为 JSON
json_config = json.dumps(config, indent=2)
配置文件示例 (config.yaml)
database: host: localhost port: 3306 username: admin password: secret123 database: myapp server: port: 8080 debug: true logging: level: INFO file: app.log
高级 YAML 解析功能
安全解析与复杂结构
import yaml
from yaml.loader import SafeLoader
class ConfigParser:
def __init__(self, config_path):
with open(config_path, 'r') as f:
self.config = yaml.load(f, Loader=SafeLoader)
def get(self, key, default=None):
"""支持点号分隔的键访问"""
keys = key.split('.')
value = self.config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
if value is None:
return default
else:
return default
return value
def merge_configs(self, *config_paths):
"""合并多个配置文件"""
merged = {}
for path in config_paths:
with open(path, 'r') as f:
config = yaml.safe_load(f)
merged.update(config)
return merged
# 使用示例
parser = ConfigParser('config.yaml')
print(parser.get('database.host')) # localhost
print(parser.get('logging.level', 'DEBUG')) # INFO
环境变量替换
import os
import re
import yaml
class EnvConfigLoader:
def __init__(self, config_path):
with open(config_path, 'r') as f:
content = f.read()
# 替换 ${VAR_NAME} 模式的环境变量
content = self._replace_env_vars(content)
self.config = yaml.safe_load(content)
def _replace_env_vars(self, content):
def replace_var(match):
env_var = match.group(1)
return os.getenv(env_var, '')
pattern = r'\$\{([^}]+)\}'
return re.sub(pattern, replace_var, content)
# 配置文件示例
"""
database:
host: ${DB_HOST}
password: ${DB_PASSWORD}
"""
Shell 脚本解析 YAML
使用 yq 工具
# 安装 yq sudo snap install yq # Linux brew install yq # macOS # 解析 YAML host=$(yq eval '.database.host' config.yaml) port=$(yq eval '.server.port' config.yaml) echo "Host: $host, Port: $port"
纯 Bash 解析（简易版）
#!/bin/bash
parse_yaml() {
local prefix=$2
local s='[[:space:]]*' w='[a-zA-Z0-9_]*' fs=$(echo @|tr @ '\034')
sed -ne "s|^\($s\)\($w\)$s:$s\"\(.*\)\"$s\$|\1$fs\2$fs\3|p" \
-e "s|^\($s\)\($w\)$s:$s\(.*\)$s\$|\1$fs\2$fs\3|p" $1 |
awk -F$fs '{
indent = length($1)/2;
vname[indent] = $2;
for (i in vname) {if (i > indent) {delete vname[i]}}
if (length($3) > 0) {
vn=""; for (i=0; i<indent; i++) {vn=(vn)(vname[i])("_")}
printf("%s%s=\"%s\"\n", "'$prefix'", vn$2, $3);
}
}'
}
# 使用
eval $(parse_yaml config.yaml)
echo $database_host # 输出: localhost
Node.js 解析 YAML
使用 js-yaml
const yaml = require('js-yaml');
const fs = require('fs');
try {
// 读取并解析
const config = yaml.load(fs.readFileSync('config.yaml', 'utf8'));
// 访问配置
console.log(config.database.host);
console.log(config.server.port);
// 验证配置
if (!config.database?.host) {
throw new Error('Missing database host configuration');
}
} catch (error) {
console.error('Failed to parse YAML:', error);
process.exit(1);
}
配置文件验证
import yaml
import jsonschema
from jsonschema import validate
# 定义配置 schema
schema = {
"type": "object",
"required": ["database", "server"],
"properties": {
"database": {
"type": "object",
"required": ["host", "port"],
"properties": {
"host": {"type": "string"},
"port": {"type": "integer", "minimum": 1, "maximum": 65535}
}
},
"server": {
"type": "object",
"properties": {
"port": {"type": "integer"},
"debug": {"type": "boolean"}
}
}
}
}
def validate_config(config_path):
with open(config_path, 'r') as f:
config = yaml.safe_load(f)
try:
validate(instance=config, schema=schema)
print("Configuration is valid!")
except jsonschema.exceptions.ValidationError as e:
print(f"Configuration error: {e.message}")
return False
return True
最佳实践建议
统一配置管理类
import yaml
from typing import Any, Dict, Optional
class ConfigurationManager:
_instance = None
_config = None
def __new__(cls):
if cls._instance is None:
cls._instance = super().__new__(cls)
return cls._instance
@classmethod
def load_config(cls, config_path: str) -> None:
with open(config_path, 'r') as f:
cls._config = yaml.safe_load(f)
@classmethod
def get(cls, key: str, default: Any = None) -> Any:
"""线程安全的配置获取"""
if cls._config is None:
raise ValueError("Configuration not loaded")
keys = key.split('.')
value = cls._config
for k in keys:
if isinstance(value, dict):
value = value.get(k)
if value is None:
return default
else:
return default
return value
@classmethod
def reload(cls, config_path: str) -> None:
"""热重载配置"""
cls.load_config(config_path)
# 使用单例模式
ConfigurationManager.load_config('config.yaml')
host = ConfigurationManager.get('database.host')
print(host)
配置文件结构建议
# config.yaml
app:
name: MyApp
version: 1.0.0
database:
host: ${DB_HOST:localhost} # 带默认值
port: ${DB_PORT:5432}
features:
- name: feature1
enabled: true
- name: feature2
enabled: false
plugins:
- name: plugin1
path: /path/to/plugin
选择合适的 YAML 解析方法取决于你的具体需求：
- Python：功能最全面，适合复杂应用
- Shell：轻量级，适合简单配置
- Node.js：适合 JavaScript 生态
- 环境变量替换：适合容器化部署
记得添加适当的错误处理和日志记录,确保配置解析的可靠性。
