如何用ruamel.yaml延迟!NETRC标签求值至url键加载完成?
问题描述
我用Python和ruamel.yaml加载配置文件,当前通过!NETRC标签从用户的.netrc文件读取密码或令牌,功能可正常运行,但部分函数需等待较长时间才尝试连接,失败体验不佳。不过求值!NETRC标签需要主机条目(对应配置中同级的url键)。配置文件示例如下:
connect: url: https://my.company/path/to/service login: mylogin token: !NETRC
我需要让token对应的!NETRC标签在url键加载完成且可访问其值后再进行求值。请问这一需求是否可行?能否通过ruamel.yaml实现?简化代码示例如下:
import ruamel.yaml yaml = ruamel.yaml.YAML(typ='safe') yaml.default_flow_style = False class NetrcTag(str): yaml_tag = '!NETRC' def __new__(cls, value): newvalue = str.__new__(cls, '!NETRC') # newvalue.netrctag = load_netrc(cfg, url value goes here) <---- newvalue.netrctag = value return newvalue @classmethod def from_yaml(cls, constructor, node): return cls(node.value) @classmethod def to_yaml(cls, represented, node): return representer.represent_scalar(cls.yaml_tag, node.netrctag)
解决方案
这个需求完全可以通过ruamel.yaml实现,核心思路是延迟!NETRC标签的求值逻辑,等整个配置结构加载完成后,再遍历配置找到!NETRC节点,结合同级的url值读取.netrc。
具体实现步骤
修改标签处理类,创建占位对象暂存标识
不再在from_yaml中直接求值,而是生成一个占位对象,标记需要后续处理的!NETRC节点:import ruamel.yaml from netrc import netrc from urllib.parse import urlparse yaml = ruamel.yaml.YAML(typ='safe') yaml.default_flow_style = False class NetrcPlaceholder: yaml_tag = '!NETRC' def __init__(self, value=None): self.value = value @classmethod def from_yaml(cls, constructor, node): return cls(node.value if node.value else None) @classmethod def to_yaml(cls, representer, node): return representer.represent_scalar(cls.yaml_tag, node.value if node.value else '') # 注册自定义标签 yaml.register_class(NetrcPlaceholder)加载配置后,递归解析占位符并替换为实际令牌
编写辅助函数遍历配置结构,找到带有NetrcPlaceholder的token字段,结合同级url解析主机名,再从.netrc中提取对应凭证:def resolve_netrc_tags(config): if isinstance(config, dict): # 检查当前节点是否有url和token,且token是占位符 if 'url' in config and 'token' in config and isinstance(config['token'], NetrcPlaceholder): url = config['url'] hostname = urlparse(url).hostname # 读取.netrc中的凭证 nrc = netrc() auth = nrc.authenticators(hostname) if auth: # auth格式为(login, account, password),此处取password作为token config['token'] = auth[2] else: raise ValueError(f".netrc中未找到主机{hostname}的对应凭证") # 递归处理子节点 for key, value in config.items(): resolve_netrc_tags(value) elif isinstance(config, list): # 递归处理列表中的每个元素 for item in config: resolve_netrc_tags(item) return config完整使用流程
# 加载配置文件 with open('config.yaml', 'r') as f: cfg = yaml.load(f) # 解析!NETRC占位符,替换为实际令牌 cfg = resolve_netrc_tags(cfg) # 后续正常使用配置即可 print(cfg['connect']['token'])
关键说明
- 此方式确保
!NETRC的求值逻辑在整个配置加载完成后执行,此时url的值已经完全可用,完美匹配需求。 - 递归函数可自动处理配置中多层级的
!NETRC节点,适配复杂结构。 - 可根据实际需求调整.netrc的读取逻辑,比如结合
login字段验证凭证匹配关系。
内容的提问来源于stack exchange,提问作者harleypig
相关产品推荐
相关产品推荐

