pythonelasticsearchprompt

How to add a user prompt to approve a list of indices before deleting them?


I have a python script where I log into ElasticSearch and delete indices that have a docs.count = 0.

To add some security to the script, I want to add a user prompt, for when the environment is Prod. It will print the list of indices and ask the user if they want to proceed with deletion. I just can't figure out where to place it or how to structure it. Can you help me please?

This is a section the current script that applies to this question:

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)

Solution

  • 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)
    

    Here's how you can modify your indices_delete function to include this feature:

    1. Add a prompt for user confirmation when the environment is "Prod". List indices to be deleted and get confirmation from the user before proceeding.
    2. Proceed with deletion only if the user confirms.