Reading custom endpoint results SDK support

dune_client/api/custom.py

@@ -0,0 +1,65 @@
+"""
+Custom endpoints API enables users to
+fetch and filter data from custom endpoints.
+"""
+
+from __future__ import annotations
+from typing import List, Optional
+
+from dune_client.api.base import BaseRouter
+from dune_client.models import (
+    DuneError,
+    ResultsResponse,
+)
+
+# pylint: disable=duplicate-code
+class CustomEndpointAPI(BaseRouter):
+    """
+    Custom endpoints API implementation.
+    Methods:
+        get_custom_endpoint_result(): returns the results of a custom endpoint.
+    """
+
+    def get_custom_endpoint_result(
+        self,
+        handle: str,
+        endpoint: str,
+        limit: Optional[int] = None,
+        offset: Optional[int] = None,
+        columns: Optional[List[str]] = None,
+        sample_count: Optional[int] = None,
+        filters: Optional[str] = None,
+        sort_by: Optional[List[str]] = None,
+    ) -> ResultsResponse:
+        """
+        Custom endpoints allow you to fetch and filter data from any 
+        custom endpoint you created.
+        More information on Custom Endpoints can be round here: 
+        https://docs.dune.com/api-reference/custom/overview
+
+        Args: 
+            handle (str): The handle of the team/user.
+            endpoint (str): The slug of the custom endpoint.
+            limit (int, optional): The maximum number of results to return.
+            offset (int, optional): The number of results to skip.
+            columns (List[str], optional): A list of columns to return.
+            sample_count (int, optional): The number of results to return.
+            filters (str, optional): The filters to apply.
+            sort_by (List[str], optional): The columns to sort by.
+        """
+        params = self._build_parameters(
+            columns=columns,
+            sample_count=sample_count,
+            filters=filters,
+            sort_by=sort_by,
+            limit=limit,
+            offset=offset,
+        )
+        response_json = self._get(
+            route=f"/endpoints/{handle}/{endpoint}/results",
+            params=params,
+        )
+        try:
+            return ResultsResponse.from_dict(response_json)
+        except KeyError as err:
+            raise DuneError(response_json, "ResultsResponse", err) from err

dune_client/api/extensions.py

@@ -20,6 +20,7 @@
 from dune_client.api.execution import ExecutionAPI
 from dune_client.api.query import QueryAPI
 from dune_client.api.table import TableAPI
+from dune_client.api.custom import CustomEndpointAPI
 from dune_client.models import (
     ResultsResponse,
     DuneError,
@@ -37,7 +38,7 @@
 POLL_FREQUENCY_SECONDS = 1
 
 
-class ExtendedAPI(ExecutionAPI, QueryAPI, TableAPI):
+class ExtendedAPI(ExecutionAPI, QueryAPI, TableAPI, CustomEndpointAPI):
     """
     Provides higher level helper methods for faster
     and easier development on top of the base ExecutionAPI.

tests/e2e/test_custom_endpoints.py

@@ -0,0 +1,23 @@
+import copy
+import os
+import time
+import unittest
+
+import dotenv
+
+from dune_client.client import DuneClient
+
+dotenv.load_dotenv()
+
+
+class TestCustomEndpoints(unittest.TestCase):
+    def setUp(self) -> None:
+        self.valid_api_key = os.environ["DUNE_API_KEY"]
+
+    def test_gettin_custom_endpoint_results(self):
+        dune = DuneClient(self.valid_api_key)
+        results = dune.get_custom_endpoint_result("dune", "new-test")
+        self.assertEqual(len(results.get_rows()), 10)
+
+if __name__ == "__main__":
+    unittest.main()