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