The
execute_query(execute)
Description:
Initiates the execution of a Dune query using the DUNE API.
Parameters:
execute (str) : The ID of the query to execute.
Returns:
Response object : Response from the Dune API with query execution details.
Copy def execute_query(execute):
headers = {"X-DUNE-API-KEY": dunekey}
url = "https://api.dune.com/api/v1/query/{execute}/execute"
try:
response = requests.post(url.format(execute=execute), headers=headers)
logging.info(response.text)
return response
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching data: {e}") wait_for_execution(execution_id, max_attempts=60, delay=5)
Description:
Monitors the status of the query execution until completion.
Parameters:
execution_id (str) : The execution ID of the query.
max_attempts (int) : Maximum number of status checks (default is 60).
delay (int) : Delay between checks in seconds (default is 5).
Returns:
Response object : The final status of the query execution.
Copy def wait_for_execution(execution_id, max_attempts=60, delay=5):
url = "https://api.dune.com/api/v1/execution/{execution_id}/status"
headers = {"X-DUNE-API-KEY": dunekey}
attempts = 0
while attempts < max_attempts:
try:
response = requests.get(url.format(execution_id=execution_id), headers=headers)
logging.info(response.text)
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching data: {e}")
if response.json().get("is_execution_finished", False):
logging.info("Execution finished!")
return response
attempts += 1
time.sleep(delay)
logging.info(f"Execution did not finish after {max_attempts} attempts.")
return None get_results(query_id)
Description:
Fetches the results of a Dune query after its execution.
Parameters:
query_id (str) : The ID of the query to retrieve results from.
Returns:
Response object : The results of the query.
Copy def get_results(query_id):
headers = {"X-DUNE-API-KEY": dunekey}
url = "https://api.dune.com/api/v1/query/{query_id}/results"
try:
response = requests.get(url.format(query_id=query_id), headers=headers)
logging.info(response.text)
return response
except requests.exceptions.RequestException as e:
logging.error(f"Error fetching data: {e}")
run_dune_query(query_id)
Description:
Runs a Dune query and retrieves the results by managing execution and result retrieval.
Parameters:
query_id (str) : The ID of the query to execute.
Returns:
dict : Results of the query in dictionary format.
Copy def run_dune_query(query_id):
try:
execution = execute_query(query_id)
execution_id = execution.json().get("execution_id")
executed_query_id = wait_for_execution(execution_id)
results = get_results(executed_query_id.json().get("query_id"))
return results.json()
except Exception as e:
logging.error(f"An error occurred: {e}")
return None Index Functions
get_alpha_index()
Description:
Fetches the data for the Alpha Index from the Dune API.
Parameters:
None.
Returns:
Copy def get_alpha_index():
return run_dune_query(query_ids["Alpha"]) get_beta_index()
Description:
Fetches the data for the Beta Index from the Dune API.
Parameters:
None.
Returns:
Copy def get_beta_index():
return run_dune_query(query_ids["Beta"]) get_gamma_index()
Description:
Fetches the data for the Gamma Index from the Dune API.
Parameters:
None.
Returns:
Copy def get_gamma_index():
return run_dune_query(query_ids["Gamma"])
Narrative Index Functions
get_daily_narrative_index()
Description:
Fetches the Daily Crypto Narrative Index (24h).
Parameters:
None.
Returns:
dict : Daily Crypto Narrative data.
Copy def get_daily_narrative_index():
return run_dune_query(query_ids["24h"]) get_weekly_narrative_index()
Description:
Fetches the Weekly Crypto Narrative Index (7d).
Parameters:
None.
Returns:
dict : Weekly Crypto Narrative data.
Copy def get_monthly_narrative_index():
return run_dune_query(query_ids["30d"]) get_monthly_narrative_index()
Description:
Fetches the Monthly Crypto Narrative Index (30d).
Parameters:
None.
Returns:
dict : Monthly Crypto Narrative data.
Copy def get_monthly_narrative_index():
return run_dune_query(query_ids["30d"])
get_quarterly_narrative_index()
Description:
Fetches the Quarterly Crypto Narrative Index (90d).
Parameters:
None.
Returns:
dict : Quarterly Crypto Narrative data.
Copy def get_quarterly_narrative_index():
return run_dune_query(query_ids["90d"])
