如何添加用户提示以在删除索引列表之前批准它们？

Question

我有一个 python 脚本，我可以在其中登录 ElasticSearch 并删除 docs.count = 0 的索引。

为了向脚本添加一些安全性，我想在环境为 Prod 时添加用户提示。它将打印索引列表并询问用户是否要继续删除。我只是不知道将其放置在哪里或如何构建它。你能帮我吗？

这是适用于该问题的当前脚本的一部分：

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 configuration

# Delete the indices with doc count 0
def indices_delete(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']}/"
  print("Actioning deletion of indices on ElasticSearch via url" , elastic_console_delete)
  for index in content:
   indiciesList = index
   collectdoccount = index['docs.count']
   search_int = int(collectdoccount)
   elasticservice_index_name = index['index']
   if search_int == 0 and not elasticservice_index_name.startswith("."):
      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("index deleted -" , "index name:" , index['index'] , ",  doc.count:" , index['docs.count'] , ",  elasticsearch url index:" , delete_url)
      if  response.status_code != 200:
          print ("index not deleted -" , "index name:" , index['index'] , ",  Reason:" , response.content)

Answer 1

def indices_delete(configuration, env):
    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)

    indices_to_delete = []
    for index in content:
        if int(index['docs.count']) == 0 and not index['index'].startswith("."):
            indices_to_delete.append(index['index'])

    if env == "Prod":
        print("Indices to be deleted:")
        for idx in indices_to_delete:
            print(idx)
        proceed = input("Do you want to proceed with the deletion of these indices? (yes/no): ")
        if proceed.lower() != "yes":
            print("Deletion aborted by user.")
            return

    elastic_console_delete = f"https://{configuration['ELASTIC_SEARCH_URL']}/"
    print("Actioning deletion of indices on ElasticSearch via url", elastic_console_delete)
    for index_name in indices_to_delete:
        delete_url = f"{elastic_console_delete}{index_name}"
        response = requests.delete(delete_url, auth=(elastic_username, elastic_password))
        if response.status_code == 200:
            print("index deleted -", "index name:", index_name, ", elasticsearch url index:", delete_url)
        else:
            print("index not deleted -", "index name:", index_name, ", Reason:", response.content)

以下是如何修改indices_delete函数以包含此功能：

1. 添加环境为“Prod”时提示用户确认。 列出要删除的索引并在继续之前获得用户的确认。
2. 仅在用户确认后才继续删除。