[WIP] pass performance param & implement "latest results" endpoint (#53)

According to API updates: Add performance param and implement latest results.

https://dune.com/docs/api/api-reference/execute-query-id/

Author: grkhr

.gitignore

@@ -6,3 +6,4 @@ _version.py
 .idea/
 venv/
 tmp/
+.vscode/

dune_client/base_client.py

@@ -20,8 +20,9 @@ class BaseDuneClient:
     API_PATH = "/api/v1"
     DEFAULT_TIMEOUT = 10
 
-    def __init__(self, api_key: str):
+    def __init__(self, api_key: str, performance: str = "medium"):
         self.token = api_key
+        self.performance = performance
         self.logger = logging.getLogger(__name__)
         logging.basicConfig(format="%(asctime)s %(levelname)s %(name)s %(message)s")
 

dune_client/client.py

@@ -7,7 +7,7 @@
 
 import time
 from io import BytesIO
-from typing import Any
+from typing import Any, Optional, Union
 
 import requests
 from requests import Response, JSONDecodeError
@@ -50,13 +50,14 @@ def _handle_response(
     def _route_url(self, route: str) -> str:
         return f"{self.BASE_URL}{self.API_PATH}/{route}"
 
-    def _get(self, route: str) -> Any:
+    def _get(self, route: str, params: Optional[Any] = None) -> Any:
         url = self._route_url(route)
         self.logger.debug(f"GET received input url={url}")
         response = requests.get(
             url,
             headers={"x-dune-api-key": self.token},
             timeout=self.DEFAULT_TIMEOUT,
+            params=params,
         )
         return self._handle_response(response)
 
@@ -71,14 +72,20 @@ def _post(self, route: str, params: Any) -> Any:
         )
         return self._handle_response(response)
 
-    def execute(self, query: Query) -> ExecutionResponse:
+    def execute(
+        self, query: Query, performance: Optional[str] = None
+    ) -> ExecutionResponse:
         """Post's to Dune API for execute `query`"""
+        self.logger.info(
+            f"executing {query.query_id} on {performance or self.performance} cluster"
+        )
         response_json = self._post(
             route=f"query/{query.query_id}/execute",
             params={
                 "query_parameters": {
                     p.key: p.to_dict()["value"] for p in query.parameters()
-                }
+                },
+                "performance": performance or self.performance,
             },
         )
         try:
@@ -122,6 +129,32 @@ def get_result_csv(self, job_id: str) -> ExecutionResultCSV:
         response.raise_for_status()
         return ExecutionResultCSV(data=BytesIO(response.content))
 
+    def get_latest_result(self, query: Union[Query, str, int]) -> ResultsResponse:
+        """
+        GET the latest results for a query_id without having to execute the query again.
+
+        :param query: :class:`Query` object OR query id as string | int  
+
+        https://dune.com/docs/api/api-reference/latest_results/    
+        """
+        if isinstance(query, Query):
+            params = {
+                f"params.{p.key}": p.to_dict()["value"] for p in query.parameters()
+            }
+            query_id = query.query_id
+        else:
+            params = None
+            query_id = query
+
+        response_json = self._get(
+            route=f"query/{query_id}/results",
+            params=params,
+        )
+        try:
+            return ResultsResponse.from_dict(response_json)
+        except KeyError as err:
+            raise DuneError(response_json, "ResultsResponse", err) from err
+
     def cancel_execution(self, job_id: str) -> bool:
         """POST Execution Cancellation to Dune API for `job_id` (aka `execution_id`)"""
         response_json = self._post(route=f"execution/{job_id}/cancel", params=None)
@@ -132,8 +165,13 @@ def cancel_execution(self, job_id: str) -> bool:
         except KeyError as err:
             raise DuneError(response_json, "CancellationResponse", err) from err
 
-    def _refresh(self, query: Query, ping_frequency: int = 5) -> str:
-        job_id = self.execute(query).execution_id
+    def _refresh(
+        self,
+        query: Query,
+        ping_frequency: int = 5,
+        performance: Optional[str] = None,
+    ) -> str:
+        job_id = self.execute(query=query, performance=performance).execution_id
         status = self.get_status(job_id)
         while status.state not in ExecutionState.terminal_states():
             self.logger.info(
@@ -147,25 +185,45 @@ def _refresh(self, query: Query, ping_frequency: int = 5) -> str:
 
         return job_id
 
-    def refresh(self, query: Query, ping_frequency: int = 5) -> ResultsResponse:
+    def refresh(
+        self,
+        query: Query,
+        ping_frequency: int = 5,
+        performance: Optional[str] = None,
+    ) -> ResultsResponse:
         """
         Executes a Dune `query`, waits until execution completes,
         fetches and returns the results.
         Sleeps `ping_frequency` seconds between each status request.
         """
-        job_id = self._refresh(query, ping_frequency=ping_frequency)
+        job_id = self._refresh(
+            query,
+            ping_frequency=ping_frequency,
+            performance=performance,
+        )
         return self.get_result(job_id)
 
-    def refresh_csv(self, query: Query, ping_frequency: int = 5) -> ExecutionResultCSV:
+    def refresh_csv(
+        self,
+        query: Query,
+        ping_frequency: int = 5,
+        performance: Optional[str] = None,
+    ) -> ExecutionResultCSV:
         """
         Executes a Dune query, waits till execution completes,
         fetches and the results in CSV format
         (use it load the data directly in pandas.from_csv() or similar frameworks)
         """
-        job_id = self._refresh(query, ping_frequency=ping_frequency)
+        job_id = self._refresh(
+            query,
+            ping_frequency=ping_frequency,
+            performance=performance,
+        )
         return self.get_result_csv(job_id)
 
-    def refresh_into_dataframe(self, query: Query) -> Any:
+    def refresh_into_dataframe(
+        self, query: Query, performance: Optional[str] = None
+    ) -> Any:
         """
         Execute a Dune Query, waits till execution completes,
         fetched and returns the result as a Pandas DataFrame
@@ -178,5 +236,5 @@ def refresh_into_dataframe(self, query: Query) -> Any:
             raise ImportError(
                 "dependency failure, pandas is required but missing"
             ) from exc
-        data = self.refresh_csv(query).data
+        data = self.refresh_csv(query, performance=performance).data
         return pandas.read_csv(data)

dune_client/client_async.py

@@ -5,7 +5,7 @@
 """
 from __future__ import annotations
 import asyncio
-from typing import Any, Optional
+from typing import Any, Optional, Union
 
 from aiohttp import (
     ClientSession,
@@ -37,13 +37,15 @@ class AsyncDuneClient(BaseDuneClient):
 
     _connection_limit = 3
 
-    def __init__(self, api_key: str, connection_limit: int = 3):
+    def __init__(
+        self, api_key: str, connection_limit: int = 3, performance: str = "medium"
+    ):
         """
         api_key - Dune API key
         connection_limit - number of parallel requests to execute.
         For non-pro accounts Dune allows only up to 3 requests but that number can be increased.
         """
-        super().__init__(api_key=api_key)
+        super().__init__(api_key=api_key, performance=performance)
         self._connection_limit = connection_limit
         self._session: Optional[ClientSession] = None
 
@@ -85,13 +87,14 @@ async def _handle_response(
             response.raise_for_status()
             raise ValueError("Unreachable since previous line raises") from err
 
-    async def _get(self, url: str) -> Any:
+    async def _get(self, url: str, params: Optional[Any] = None) -> Any:
         if self._session is None:
             raise ValueError("Client is not connected; call `await cl.connect()`")
         self.logger.debug(f"GET received input url={url}")
         response = await self._session.get(
             url=f"{self.API_PATH}{url}",
             headers=self.default_headers(),
+            params=params,
         )
         return await self._handle_response(response)
 
@@ -106,11 +109,19 @@ async def _post(self, url: str, params: Any) -> Any:
         )
         return await self._handle_response(response)
 
-    async def execute(self, query: Query) -> ExecutionResponse:
+    async def execute(
+        self, query: Query, performance: Optional[str] = None
+    ) -> ExecutionResponse:
         """Post's to Dune API for execute `query`"""
+        params = query.request_format()
+        params["performance"] = performance or self.performance
+
+        self.logger.info(
+            f"executing {query.query_id} on {performance or self.performance} cluster"
+        )
         response_json = await self._post(
             url=f"/query/{query.query_id}/execute",
-            params=query.request_format(),
+            params=params,
         )
         try:
             return ExecutionResponse.from_dict(response_json)
@@ -135,6 +146,30 @@ async def get_result(self, job_id: str) -> ResultsResponse:
         except KeyError as err:
             raise DuneError(response_json, "ResultsResponse", err) from err
 
+    async def get_latest_result(self, query: Union[Query, str, int]) -> ResultsResponse:
+        """
+        GET the latest results for a query_id without having to execute the query again.
+
+        https://dune.com/docs/api/api-reference/latest_results/
+        """
+        if isinstance(query, Query):
+            params = {
+                f"params.{p.key}": p.to_dict()["value"] for p in query.parameters()
+            }
+            query_id = query.query_id
+        else:
+            params = None
+            query_id = query
+
+        response_json = await self._get(
+            url=f"/query/{query_id}/results",
+            params=params,
+        )
+        try:
+            return ResultsResponse.from_dict(response_json)
+        except KeyError as err:
+            raise DuneError(response_json, "ResultsResponse", err) from err
+
     async def cancel_execution(self, job_id: str) -> bool:
         """POST Execution Cancellation to Dune API for `job_id` (aka `execution_id`)"""
         response_json = await self._post(url=f"/execution/{job_id}/cancel", params=None)
@@ -145,13 +180,18 @@ async def cancel_execution(self, job_id: str) -> bool:
         except KeyError as err:
             raise DuneError(response_json, "CancellationResponse", err) from err
 
-    async def refresh(self, query: Query, ping_frequency: int = 5) -> ResultsResponse:
+    async def refresh(
+        self,
+        query: Query,
+        ping_frequency: int = 5,
+        performance: Optional[str] = None,
+    ) -> ResultsResponse:
         """
         Executes a Dune `query`, waits until execution completes,
         fetches and returns the results.
         Sleeps `ping_frequency` seconds between each status request.
         """
-        job_id = (await self.execute(query)).execution_id
+        job_id = (await self.execute(query=query, performance=performance)).execution_id
         status = await self.get_status(job_id)
         while status.state not in ExecutionState.terminal_states():
             self.logger.info(

dune_client/query.py

@@ -3,7 +3,7 @@
 """
 import urllib.parse
 from dataclasses import dataclass
-from typing import Optional, List, Dict
+from typing import Optional, List, Dict, Union
 
 from dune_client.types import QueryParameter
 
@@ -41,7 +41,7 @@ def __hash__(self) -> int:
         """
         return self.url().__hash__()
 
-    def request_format(self) -> Dict[str, Dict[str, str]]:
+    def request_format(self) -> Dict[str, Union[Dict[str, str], str, None]]:
         """Transforms Query objects to params to pass in API"""
         return {
             "query_parameters": {p.key: p.to_dict()["value"] for p in self.parameters()}