Too many positional arguments

.gitignore

@@ -5,6 +5,7 @@ dist
 _version.py
 .idea/
 venv/
+.venv/
 tmp/
 .vscode/
 build/

.pylintrc

@@ -1,4 +1,5 @@
 [MASTER]
-disable=fixme,logging-fstring-interpolation,too-many-positional-arguments
+disable=fixme,logging-fstring-interpolation
 [DESIGN]
 max-args=10
+max-attributes=7

dune_client/api/base.py

@@ -9,7 +9,7 @@
 import logging.config
 import os
 from json import JSONDecodeError
-from typing import Any, Dict, List, Optional, Union, IO
+from typing import Any, Dict, Optional, IO
 
 from requests import Response, Session
 from requests.adapters import HTTPAdapter, Retry
@@ -30,7 +30,7 @@ class BaseDuneClient:
     and provides some convenient functions to use in other clients
     """
 
-    def __init__(  # pylint: disable=too-many-arguments
+    def __init__(  # pylint: disable=too-many-arguments, too-many-positional-arguments
         self,
         api_key: str,
         base_url: str = "https://api.dune.com",
@@ -83,51 +83,6 @@ def default_headers(self) -> Dict[str, str]:
             "User-Agent": f"dune-client/{client_version} (https://pypi.org/project/dune-client/)",
         }
 
-    ############
-    # Utilities:
-    ############
-
-    def _build_parameters(
-        self,
-        params: Optional[Dict[str, Union[str, int]]] = None,
-        columns: Optional[List[str]] = None,
-        sample_count: Optional[int] = None,
-        filters: Optional[str] = None,
-        sort_by: Optional[List[str]] = None,
-        limit: Optional[int] = None,
-        offset: Optional[int] = None,
-        allow_partial_results: str = "true",
-    ) -> Dict[str, Union[str, int]]:
-        """
-        Utility function that builds a dictionary of parameters to be used
-        when retrieving advanced results (filters, pagination, sorting, etc.).
-        This is shared between the sync and async client.
-        """
-        # Ensure we don't specify parameters that are incompatible:
-        assert (
-            # We are not sampling
-            sample_count is None
-            # We are sampling and don't use filters or pagination
-            or (limit is None and offset is None and filters is None)
-        ), "sampling cannot be combined with filters or pagination"
-
-        params = params or {}
-        params["allow_partial_results"] = allow_partial_results
-        if columns is not None and len(columns) > 0:
-            params["columns"] = ",".join(columns)
-        if sample_count is not None:
-            params["sample_count"] = sample_count
-        if filters is not None:
-            params["filters"] = filters
-        if sort_by is not None and len(sort_by) > 0:
-            params["sort_by"] = ",".join(sort_by)
-        if limit is not None:
-            params["limit"] = limit
-        if offset is not None:
-            params["offset"] = offset
-
-        return params
-
 
 class BaseRouter(BaseDuneClient):
     """Extending the Base Client with elementary api routing"""

dune_client/api/custom.py

@@ -4,7 +4,7 @@
 """
 
 from __future__ import annotations
-from typing import List, Optional
+from typing import List, NamedTuple, Optional
 
 from dune_client.api.base import BaseRouter
 from dune_client.models import (
@@ -13,6 +13,19 @@
 )
 
 
+class CustomAPIParams(NamedTuple):
+    """
+    Params for Custom Endpoint API Function
+    """
+
+    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
+
+
 # pylint: disable=duplicate-code
 class CustomEndpointAPI(BaseRouter):
     """
@@ -25,12 +38,7 @@ 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,
+        params: Optional[CustomAPIParams] = None,
     ) -> ResultsResponse:
         """
         Custom endpoints allow you to fetch and filter data from any
@@ -48,17 +56,12 @@ def get_custom_endpoint_result(
             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,
-        )
+        if params is None:
+            params = CustomAPIParams()
+
         response_json = self._get(
             route=f"/endpoints/{handle}/{endpoint}/results",
-            params=params,
+            params=params._asdict(),
         )
         try:
             return ResultsResponse.from_dict(response_json)

dune_client/api/execution.py

@@ -7,7 +7,7 @@
 """
 
 from io import BytesIO
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, NamedTuple, Optional
 
 from deprecated import deprecated
 
@@ -27,6 +27,21 @@
 from dune_client.query import QueryBase
 
 
+class GetExecutionResultsParams(NamedTuple):
+    """
+    Parameters for get execution result functions
+    """
+
+    limit: Optional[int] = None
+    columns: Optional[List[str]] = None
+    batch_size: Optional[int] = None
+    sample_count: Optional[int] = None
+    filters: Optional[str] = None
+    sort_by: Optional[List[str]] = None
+    offset: Optional[int] = None
+    allow_partial_results: str = "true"
+
+
 class ExecutionAPI(BaseRouter):
     """
     Query execution and result fetching functions.
@@ -75,38 +90,19 @@ def get_execution_status(self, job_id: str) -> ExecutionStatusResponse:
     def get_execution_results(
         self,
         job_id: 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,
-        allow_partial_results: str = "true",
+        params: Optional[GetExecutionResultsParams] = None,
     ) -> ResultsResponse:
         """GET results from Dune API for `job_id` (aka `execution_id`)"""
-        params = self._build_parameters(
-            columns=columns,
-            sample_count=sample_count,
-            filters=filters,
-            sort_by=sort_by,
-            limit=limit,
-            offset=offset,
-            allow_partial_results=allow_partial_results,
-        )
+
+        if params is None:
+            params = GetExecutionResultsParams()
 
         route = f"/execution/{job_id}/results"
         url = self._route_url(route)
-        return self._get_execution_results_by_url(url=url, params=params)
+        return self._get_execution_results_by_url(url=url, params=params._asdict())
 
     def get_execution_results_csv(
-        self,
-        job_id: str,
-        limit: Optional[int] = None,
-        offset: Optional[int] = None,
-        columns: Optional[List[str]] = None,
-        filters: Optional[str] = None,
-        sample_count: Optional[int] = None,
-        sort_by: Optional[List[str]] = None,
+        self, job_id: str, params: Optional[GetExecutionResultsParams] = None
     ) -> ExecutionResultCSV:
         """
         GET results in CSV format from Dune API for `job_id` (aka `execution_id`)
@@ -115,18 +111,13 @@ def get_execution_results_csv(
         use this method for large results where you want lower CPU and memory overhead
         if you need metadata information use get_results() or get_status()
         """
-        params = self._build_parameters(
-            columns=columns,
-            sample_count=sample_count,
-            filters=filters,
-            sort_by=sort_by,
-            limit=limit,
-            offset=offset,
-        )
+
+        if params is None:
+            params = GetExecutionResultsParams()
 
         route = f"/execution/{job_id}/results/csv"
         url = self._route_url(route)
-        return self._get_execution_results_csv_by_url(url=url, params=params)
+        return self._get_execution_results_csv_by_url(url=url, params=params._asdict())
 
     def _get_execution_results_by_url(
         self, url: str, params: Optional[Dict[str, Any]] = None