如何将Python脚本中environments函数循环内的变量传递到其他函数

原脚本核心问题 #

• 局部变量作用域限制：ELASTIC_SEARCH_USER/ELASTIC_SEARCH_PASSWORD/ELASTIC_SEARCH_URL是environments函数内的局部变量，外部函数无法直接访问。
• 无效循环逻辑：for configuration in _secret_client完全冗余，既未实现遍历密钥的需求，还会覆盖configuration变量，导致返回值不符合预期。
• 参数传递错误：主函数中pass_credentials是environments返回的元组（Key Vault URL + 错误的configuration），与indice_delete的参数要求不匹配。

修改后的完整代码 #

from azure.identity import DefaultAzureCredential
from azure.keyvault.secrets import SecretClient
import argparse
import requests
import json

CREDENTIAL = DefaultAzureCredential(exclude_shared_token_cache_credential=True, exclude_environment_credential=True, exclude_managed_identity_credential=True, exclude_visual_studio_code_credential=True)
KV_PARTIAL_NAME = "KeyVault_name-"

def get_secret(kv_client, secret_name):
    return kv_client.get_secret(secret_name).value 

# Subscription identification    
def environments():
    args = parse_arguments() 
    env = args.env_name
    subscription_name = env if env != "Prod" else "Production"
    print(f"The script is running on {subscription_name}")
    
    azure_key_vault_env_specific_url = f"https://{KV_PARTIAL_NAME}-{env}.vault.azure.net/"
    _secret_client = SecretClient(vault_url=azure_key_vault_env_specific_url, credential=CREDENTIAL)
    
    # 直接获取所需密钥并封装为字典
    configuration = {
        "elastic_user": _secret_client.get_secret("Elastic--User").value,
        "elastic_password": _secret_client.get_secret("Elastic--Password").value,
        "elastic_url": _secret_client.get_secret("Elastic--URL").value
    }
    return azure_key_vault_env_specific_url, configuration

# Provide argument -e for the script to choose the correct subscription
def parse_arguments():
    parser = argparse.ArgumentParser()
    parser.add_argument('--env_name', '-e', type=str, choices=['Dev', 'QA', 'Prod'],  help='The short for the environment for subscription', required=True)
    return parser.parse_args()

def indice_delete(configuration):
    # 从传入的字典中读取凭据
    elastic_user = configuration["elastic_user"]
    elastic_password = configuration["elastic_password"]
    elastic_url = configuration["elastic_url"]
    
    elastic_auth_uri = f"https://{elastic_url}/security/_authenticate"
    response = requests.get(elastic_auth_uri, auth=(elastic_user, elastic_password))        
    search_url_index = "_cat/indices/"
    params_dict = {
        "h":"index,docs.count",
        "s":"docs.count:asc",
        "format":"json"
    }
    elastic_console = f"https://{elastic_url}/{search_url_index}"
    getRequestElasticSearch = requests.get(elastic_console, auth=(elastic_user, elastic_password), params=params_dict)
    content = json.loads(getRequestElasticSearch.text)
    elastic_console_delete = f"https://{elastic_url}/"
    
    for index in content:
        collectdoccount = index['docs.count']
        search_int = int(collectdoccount)
        if search_int == 0:
            index_name = index['index']
            delete_url = f"{elastic_console_delete}{index_name}"
            response = requests.delete(delete_url, auth=(elastic_user, elastic_password))
            if response.status_code == 200:
                print(f"index deleted - index name: {index['index']}, doc.count: {index['docs.count']}, elasticsearch url index: {delete_url}")
            else:
                print(f"index not deleted - index name: {index['index']}, Reason: {response.content}")

if __name__ == '__main__':
    # 解构返回的元组，只传递configuration给indice_delete
    _, config = environments()
    indice_delete(config)

关键修改说明 #

1. 凭据封装：将三个Elasticsearch凭据封装为字典返回，确保作用域可传递。
2. 参数传递修正：主函数中解构environments的返回值，只把凭据字典传给indice_delete。
3. 消除无效循环：移除无意义的for configuration in _secret_client循环，直接获取指定密钥。
4. 变量本地化：在indice_delete内从传入的字典读取凭据，避免依赖外部未定义变量。

内容的提问来源于stack exchange，提问作者JonH