如何在删除ElasticSearch空索引前添加生产环境用户确认提示

1. 调整配置函数，传递环境标识 #

先修改get_Configuration函数，让它同时返回环境标识和配置信息，这样后续删除逻辑能判断当前是否为生产环境：

def get_Configuration():
    env = "Dev"
    subscription_name = env
    if env == "Prod":
        subscription_name = "Production"
    print(f"The script is running on {subscription_name}")
    KV_name_entry = f"Elastic--elastic"
    _secret_client = SecretClient(vault_url=f"https://{KV_PARTIAL_NAME}-{env}.vault.azure.net/", credential=CREDENTIAL)
    configuration = {
        'ELASTIC_SEARCH_USER': KV_name_entry.split("--")[-1],
        'ELASTIC_SEARCH_PASSWORD' : _secret_client.get_secret(KV_name_entry).value,
        'ELASTIC_SEARCH_URL' : _secret_client.get_secret("Elastic--URL").value
    }
    # 同时返回环境标识和配置
    return env, configuration

2. 修改删除函数，添加生产环境确认逻辑 #

重构indices_delete函数，先收集所有待删除的索引，再根据环境决定是否需要用户确认：

# Delete the indices with doc count 0
def indices_delete(env, configuration):
    elastic_auth_uri = f"https://{configuration['ELASTIC_SEARCH_URL']}/security/_authenticate"
    elastic_username = configuration['ELASTIC_SEARCH_USER']
    elastic_password = configuration['ELASTIC_SEARCH_PASSWORD']
    authenticate_response = requests.get(elastic_auth_uri, auth=(elastic_username, elastic_password))
    if authenticate_response.status_code == 405:
        print(f"User: {elastic_username} has been authenticated.")
    
    search_url_index = "_cat/indices/"
    params_dict = {
        "h":"index,docs.count",
        "s":"docs.count:asc",
        "format":"json"
    }
    elastic_console = f"https://{configuration['ELASTIC_SEARCH_URL']}/{search_url_index}"
    getRequestElasticSearch = requests.get(elastic_console, auth=(elastic_username, elastic_password), params=params_dict)
    content = json.loads(getRequestElasticSearch.text)
    elastic_console_delete = f"https://{configuration['ELASTIC_SEARCH_URL']}/"
    
    # 先收集所有待删除的索引
    indices_to_delete = []
    for index in content:
        collectdoccount = index['docs.count']
        search_int = int(collectdoccount)
        elasticservice_index_name = index['index']
        if search_int == 0 and not elasticservice_index_name.startswith("."):
            indices_to_delete.append(index)
    
    # 处理待删除索引
    if not indices_to_delete:
        print("没有符合条件的索引需要删除")
        return
    
    print(f"发现 {len(indices_to_delete)} 个待删除索引：")
    for idx, index in enumerate(indices_to_delete, 1):
        print(f"{idx}. 索引名: {index['index']}, 文档数: {index['docs.count']}")
    
    # 生产环境下要求用户确认
    if env == "Prod":
        confirm = input("确认要删除以上索引吗？输入 'Y' 继续，其他输入取消：").strip().upper()
        if confirm != "Y":
            print("删除操作已取消")
            return
    
    # 执行删除操作
    print("开始执行删除操作...")
    for index in indices_to_delete:
        index_name = index['index']
        delete_url = f"{elastic_console_delete}{index_name}"
        response = requests.delete(delete_url, auth=(elastic_username, elastic_password))
        if response.status_code == 200:
            print(f"已删除索引 - 索引名: {index['index']}, 文档数: {index['docs.count']}, URL: {delete_url}")
        else:
            print(f"删除失败 - 索引名: {index['index']}, 原因: {response.content}")

3. 调整函数调用逻辑 #

调用时需要同时接收环境变量和配置：

# 主逻辑调用
env, config = get_Configuration()
indices_delete(env, config)

关键逻辑说明 #

• 预收集待删除索引：先筛选出所有符合条件的索引，避免边查边删导致的状态不一致问题
• 生产环境强校验：仅在生产环境触发用户交互，确保操作的安全性
• 清晰的操作反馈：展示待删除列表，让用户明确操作对象，取消操作直接终止流程

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