TypeError: indice_delete() takes 0 positional arguments but 3 were given

22 views Asked by At

I am a little lost on how to proceed with my script.

There are two functions (environments and indice_delete) that I'm stuck with. So the environments functions provides all the necessary ElasticSearch credential login to allow the indice_delete function to gain access to ElasticSearch to pursue deleting specific indices.

Initially I had issues passing the credential variables (ELASTIC_SEARCH_USER, ELASTIC_SEARCH_PASSWORD, ELASTIC_SEARCH_URL) from environments(). But I have made these variables global.

However I am getting the error: TypeError: indice_delete() takes 0 positional arguments but 3 were given

My script is below:

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 = "KV-"
    
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":
     subscription_name = "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)
  global ELASTIC_SEARCH_USER 
  ELASTIC_SEARCH_USER = _secret_client.get_secret("Elastic--User").value
  global ELASTIC_SEARCH_PASSWORD
  ELASTIC_SEARCH_PASSWORD = _secret_client.get_secret("Elastic--Password").value
  global ELASTIC_SEARCH_URL
  ELASTIC_SEARCH_URL = _secret_client.get_secret("Elastic--URL").value
  return azure_key_vault_env_specific_url, ELASTIC_SEARCH_USER,ELASTIC_SEARCH_PASSWORD,ELASTIC_SEARCH_URL

# 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():
  elastic_auth_uri = f"https://{ELASTIC_SEARCH_URL}/security/_authenticate"
  response = requests.get(elastic_auth_uri, auth=(ELASTIC_SEARCH_USER, ELASTIC_SEARCH_PASSWORD))        
  search_url_index = "_cat/indices/"
  params_dict = {
  "h":"index,docs.count",
  "s":"docs.count:asc",
  "format":"json"
  }
  elastic_console = f"https://{ELASTIC_SEARCH_URL}/{search_url_index}"
  getRequestElasticSearch = requests.get(elastic_console, auth=(ELASTIC_SEARCH_USER, ELASTIC_SEARCH_PASSWORD), params=params_dict)
  content = json.loads(getRequestElasticSearch.text)
  elastic_console_delete = f"https://{ELASTIC_SEARCH_URL}/"
  for index in content:
   indiciesList = index
   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_SEARCH_USER, ELASTIC_SEARCH_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)

if __name__ == '__main__':
   environments()
   indice_delete(ELASTIC_SEARCH_USER, ELASTIC_SEARCH_PASSWORD, ELASTIC_SEARCH_URL)
0

There are 0 answers