TypeError: indice_delete() 接收了 0 个位置参数但给了 3 个

-1 投票
0 回答
20 浏览
提问于 2025-04-12 05:03

我对我的脚本有点迷茫,不知道该怎么继续。

我卡在两个函数上,分别是 environmentsindice_delete

这个 environments 函数提供了所有必要的 ElasticSearch 登录凭证,以便 indice_delete 函数能够访问 ElasticSearch,进行特定索引的删除。

最开始我在传递凭证变量(ELASTIC_SEARCH_USERELASTIC_SEARCH_PASSWORDELASTIC_SEARCH_URL)时遇到了问题,因为这些变量是在 environments() 函数里定义的。不过我已经把这些变量设置成全局的了。

但是我现在遇到了一个错误:TypeError: indice_delete() takes 0 positional arguments but 3 were given,意思是 indice_delete 函数不接受任何位置参数,但我给了它三个。

我的脚本如下:

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 个回答

暂无回答

撰写回答