importosimportrefromfunctoolsimportpartialimportyamlhas_regex_module=FalseENV_VAR_MATCHER=re.compile(r"""\$\{#按字面意思匹配字符`${`([^}:\s]+)#第一组：匹配除`}之外的任何字符`or`:`:?#匹配文字`:`字符零次或一次([^}]+)?#第二组：匹配除`}`之外的任何字符\}#匹配字符`}`字面意思""",re.VERBOSE)IMPLICIT_ENV_VAR_MATCHER=re.compile(r""".*#匹配任意数量的任意字符\$\{.*\}#匹配任意数量的任意字符#在`${`和`}`之间.*#匹配任意数量的任意字符""",re.VERBOSE)RECURSIVE_ENV_VAR_MATCHER=re.compile(r"""\$\{#字面上匹配字符`${`([^}]+)?#匹配任意字符exept`}`\}#按字面意思匹配字符`}`([^$}]+)?#匹配除`}`或`$`之外的任何字符\}#匹配字符`}`字面意思""",re.VERBOSE,)def_replace_env_var(match):env_var,default=match.groups()value=os.environ.get(env_var,None)ifvalueisNone:#expanddefaultusingothervarsifdefaultisNone:#regexmodulereturnNone而不是#''ifenginedidn'tentereddefaultcapturegroupdefault=''value=defaultwhileIMPLICIT_ENV_VAR_MATCHER.match(value):#pragma:nocovervalue=ENV_VAR_MATCHER.sub(_replace_env_var,value)returnvaluedefenv_var_constructor(loader,node,raw=False):raw_value=loader.construct_scalar(node)#在递归环境中检测错误变量如果没有has_regex_module和RECURSIVE_ENV_VAR_MATCHER.match(raw_value):#pragma:nocoverraiseException("Nestedenvironmentvariablelookuprequiresthe`regex`module")value=ENV_VAR_MATCHER.sub(_replace_env_var,raw_value)ifvalue==raw_value:returnvalue#避免递归返回值ifrawelseyaml.safe_load(值）defsetup_yaml_parser（）：yaml.add_constructor（'！env_var'，env_var_constructor，yaml.SafeLoader）yaml.add_constructor（'！raw_env_var'，部分（env_var_constructor，raw=True），yaml.SafeLoader）yaml.add_implicit_resolver（'！env_var',IMPLICIT_ENV_VAR_MATCHER,Loader=yaml.SafeLoader)#copyfromnamekohttps://github.com/nameko/nameko/blob/v2.14.1/nameko/cli/main.py写个元测试#coding=utf-8importcsvimportjsonimportosimporttimeimportunittestfromcollectionsimportnamedtuplefromconcurrent.futuresimportThreadPoolExecutorfrompathlibimportPathfromuuidimportuuid4#fromelasticsearchimportElasticsearchfromloguruimportloggerimportsettingsfrommarkimportBASE_DIRfrompathlibimportPathimportyamlTESTING_BASE_DIR=Path(__file__).resolve().parentclassTestEnvParse(unittest.TestCase):deftest_parse_yaml_with_env(self):"""测试es是否联通python-munittesttesting.test_env.TestEnvParse.test_parse_yaml_with_env"""从pon.events导入EventletEventRunnerconfig_filepath=TESTING_BASE_DIR/'config.yaml'withopen(config_filepath,'r',encoding='utf-8')asf:config:dict[str,dict]=yaml.safe_load(f)logger.debug(config)准备的yaml文件AMQP_URI:amqp://${RABBIT_USER:guest}:${RABBIT_PASSWORD:guest}@${RABBIT_HOST:localhost}:${RABBIT_PORT:5672}/${RABBIT_VHOST:/}prd:database:mysql:host:${MYSQL_HOST:110.110.100.110}port:3306username:you密码:youdatabase_name:haha??elasticsearch:主机:1110.110.110.100端口:9200用户名:hah密码:hahindex_name:hehe运行命令:MYSQL_HOST=192.168.31.245python-munittesttesting.test_env.TestEnvParse.test_parse_yaml_with_env2022-09-247|DEB.29:13:testing.test_env:test_parse_yaml_with_env:33-{'AMQP_URI':'amqp://pon:pon@124.222.178.120:5672//','prd':{'数据库':{'mysql':{'host':'192.168.31.245','port':3306,'username':'you','password':'you','database_name':'haha'},'elasticsearch':{'host':'1110.110.110.100','port':9200,'username':'hah','password':'hah','index_name':'hehe'}}}}运行结果，可以看到mysql主机已被环境变量替换
