169 lines
5.9 KiB
Python
169 lines
5.9 KiB
Python
#!/usr/bin/env python
|
|
# -*- coding: utf-8 -*-
|
|
# @Time : 2025/3/15 17:21
|
|
# @Author : AngesZhu
|
|
# @File : yaml_utils.py
|
|
# @Desc : yaml操作封装类
|
|
import yaml
|
|
import os
|
|
from typing import Any, Dict
|
|
|
|
|
|
class YAMLOperator:
|
|
"""
|
|
一个用于操作 YAML 文件的工具类。
|
|
支持读取、写入、更新和查询 YAML 文件内容。
|
|
"""
|
|
|
|
def __init__(self, file_path: str):
|
|
"""
|
|
初始化 YAMLOperator 实例。
|
|
:param file_path: YAML 文件的路径。
|
|
"""
|
|
self.file_path = file_path
|
|
self.data = self._load_yaml()
|
|
|
|
def _load_yaml(self) -> Dict[str, Any]:
|
|
"""
|
|
加载 YAML 文件并解析为字典。
|
|
:return: 解析后的字典。
|
|
:raises FileNotFoundError: 如果文件不存在。
|
|
:raises yaml.YAMLError: 如果 YAML 格式错误。
|
|
"""
|
|
try:
|
|
if not os.path.exists(self.file_path):
|
|
raise FileNotFoundError(f"文件 '{self.file_path}' 不存在")
|
|
|
|
with open(self.file_path, "r", encoding="utf-8") as file:
|
|
return yaml.safe_load(file) or {}
|
|
except FileNotFoundError as e:
|
|
raise FileNotFoundError(f"文件未找到: {e}")
|
|
except yaml.YAMLError as e:
|
|
raise ValueError(f"YAML 格式错误: {e}")
|
|
except Exception as e:
|
|
raise RuntimeError(f"加载 YAML 文件时发生未知错误: {e}")
|
|
|
|
def get_data(self):
|
|
"""获取已经读取的文件内容"""
|
|
return self.data
|
|
|
|
def save_yaml(self):
|
|
"""
|
|
将当前数据保存到 YAML 文件中。
|
|
"""
|
|
try:
|
|
with open(self.file_path, "w", encoding="utf-8") as file:
|
|
yaml.dump(self.data, file, allow_unicode=True, default_flow_style=False)
|
|
except PermissionError:
|
|
raise PermissionError(f"无法写入文件 '{self.file_path}'，请检查权限")
|
|
except Exception as e:
|
|
raise RuntimeError(f"保存 YAML 文件时发生未知错误: {e}")
|
|
|
|
def get_value(self, key_path: str, default: Any = None) -> Any:
|
|
"""
|
|
根据键路径获取嵌套字典中的值。
|
|
:param key_path: 键路径，用点号分隔（如 'key.subkey'）。
|
|
:param default: 如果键不存在时返回的默认值。
|
|
:return: 查询到的值或默认值。
|
|
"""
|
|
try:
|
|
if not isinstance(key_path, str):
|
|
raise ValueError("key_path 必须是一个字符串")
|
|
keys = key_path.split(".")
|
|
current = self.data
|
|
for key in keys:
|
|
if isinstance(current, dict) and key in current:
|
|
current = current[key]
|
|
else:
|
|
return default
|
|
return current
|
|
except Exception as e:
|
|
raise RuntimeError(f"获取 YAML 值时发生错误: {e}")
|
|
|
|
def set_value(self, key_path: str, value: Any):
|
|
"""
|
|
根据键路径设置嵌套字典中的值。
|
|
如果中间层级不存在，则自动创建。
|
|
:param key_path: 键路径，用点号分隔（如 'key.subkey'）。
|
|
:param value: 要设置的值。
|
|
"""
|
|
try:
|
|
if not isinstance(key_path, str):
|
|
raise ValueError("key_path 必须是一个字符串")
|
|
keys = key_path.split(".")
|
|
current = self.data
|
|
for key in keys[:-1]:
|
|
if key not in current or not isinstance(current[key], dict):
|
|
current[key] = {}
|
|
current = current[key]
|
|
current[keys[-1]] = value
|
|
except Exception as e:
|
|
raise RuntimeError(f"设置 YAML 值时发生错误: {e}")
|
|
|
|
def update_data(self, new_data: Dict[str, Any]):
|
|
"""
|
|
更新整个 YAML 数据。
|
|
:param new_data: 新的字典数据。
|
|
"""
|
|
try:
|
|
if not isinstance(new_data, dict):
|
|
raise ValueError("new_data 必须是一个字典")
|
|
self.data.update(new_data)
|
|
except Exception as e:
|
|
raise RuntimeError(f"更新 YAML 数据时发生错误: {e}")
|
|
|
|
def delete_key(self, key_path: str):
|
|
"""
|
|
根据键路径删除嵌套字典中的键。
|
|
:param key_path: 键路径，用点号分隔（如 'key.subkey'）。
|
|
:raises KeyError: 如果键不存在。
|
|
"""
|
|
try:
|
|
if not isinstance(key_path, str):
|
|
raise ValueError("key_path 必须是一个字符串")
|
|
keys = key_path.split(".")
|
|
current = self.data
|
|
for key in keys[:-1]:
|
|
if key in current and isinstance(current[key], dict):
|
|
current = current[key]
|
|
else:
|
|
raise KeyError(f"键路径 '{key_path}' 不存在")
|
|
if keys[-1] in current:
|
|
del current[keys[-1]]
|
|
else:
|
|
raise KeyError(f"键路径 '{key_path}' 不存在")
|
|
except Exception as e:
|
|
raise RuntimeError(f"删除 YAML 键时发生错误: {e}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
# 初始化 YAML 操作类
|
|
yaml_operator = YAMLOperator("config.yaml")
|
|
|
|
# 查询嵌套数据
|
|
print(yaml_operator.get_value("database.host")) # 输出: localhost
|
|
|
|
# 设置新值
|
|
yaml_operator.set_value("database.port", 5432)
|
|
|
|
# 删除键
|
|
yaml_operator.delete_key("logging.level")
|
|
|
|
# 保存更改到文件
|
|
yaml_operator.save_yaml()
|
|
|
|
# 处理异常情况
|
|
yaml_operator.get_value("invalid.key") # 返回默认值 None
|
|
yaml_operator.delete_key("nonexistent.key") # 抛出 KeyError
|
|
except FileNotFoundError as e:
|
|
print(f"文件未找到: {e}")
|
|
except PermissionError as e:
|
|
print(f"权限错误: {e}")
|
|
except ValueError as e:
|
|
print(f"输入错误: {e}")
|
|
except KeyError as e:
|
|
print(f"键错误: {e}")
|
|
except RuntimeError as e:
|
|
print(f"运行时错误: {e}")
|