Too many positional arguments

.gitignore

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

.pylintrc

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

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, Union, IO
 
 from requests import Response, Session
 from requests.adapters import HTTPAdapter, Retry
@@ -89,44 +89,47 @@ def default_headers(self) -> Dict[str, str]:
 
     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",
+        params: Optional[Dict[str, Any]] = None,
     ) -> 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:
+        if params is None:
+            params = {}
+        parameters = params.get("params", None)
+        columns = params.get("columns", None)
+        sample_count = params.get("sample_count", None)
+        filters = params.get("filters", None)
+        sort_by = params.get("sort_by", None)
+        limit = params.get("limit", None)
+        offset = params.get("offset", None)
         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
+        parameters = parameters or {}
+        parameters["allow_partial_results"] = allow_partial_results
         if columns is not None and len(columns) > 0:
-            params["columns"] = ",".join(columns)
+            parameters["columns"] = ",".join(columns)
         if sample_count is not None:
-            params["sample_count"] = sample_count
+            parameters["sample_count"] = sample_count
         if filters is not None:
-            params["filters"] = filters
+            parameters["filters"] = filters
         if sort_by is not None and len(sort_by) > 0:
-            params["sort_by"] = ",".join(sort_by)
+            parameters["sort_by"] = ",".join(sort_by)
         if limit is not None:
-            params["limit"] = limit
+            parameters["limit"] = limit
         if offset is not None:
-            params["offset"] = offset
+            parameters["offset"] = offset
 
-        return params
+        return parameters
 
 
 class BaseRouter(BaseDuneClient):

dune_client/api/custom.py

@@ -4,7 +4,7 @@
 """
 
 from __future__ import annotations
-from typing import List, Optional
+from typing import Dict, Optional, Any
 
 from dune_client.api.base import BaseRouter
 from dune_client.models import (
@@ -25,12 +25,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[Dict[str, Any]] = None,
     ) -> ResultsResponse:
         """
         Custom endpoints allow you to fetch and filter data from any
@@ -48,17 +43,28 @@ 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 = {}
+        limit = params.get("limit", None)
+        offset = params.get("offset", None)
+        columns = params.get("columns", None)
+        sample_count = params.get("sample_counts", None)
+        filters = params.get("filters", None)
+        sort_by = params.get("sort_by", None)
+
+        build_params = self._build_parameters(
+            params={
+                "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,
+            params=build_params,
         )
         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,19 @@
 from dune_client.query import QueryBase
 
 
+class GetExecutionResultsParams(NamedTuple):
+    """
+    Parameters for get execution result functions
+    """
+
+    limit: Optional[int] = None
+    columns: Optional[List[str]] = None
+    sample_count: Optional[int] = None
+    filters: Optional[str] = None
+    sort_by: Optional[List[str]] = None
+    offset: Optional[int] = None
+
+
 class ExecutionAPI(BaseRouter):
     """
     Query execution and result fetching functions.
@@ -75,38 +88,30 @@ 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,
+        params: Optional[GetExecutionResultsParams] = None,
         allow_partial_results: str = "true",
     ) -> 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,
-        )
+        build_params = None
+        if params is not None:
+            build_params = self._build_parameters(
+                allow_partial_results=allow_partial_results,
+                params={
+                    "columns": params.columns,
+                    "sample_count": params.sample_count,
+                    "filters": params.filters,
+                    "sort_by": params.sort_by,
+                    "limit": params.limit,
+                    "offset": params.offset,
+                },
+            )
 
         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=build_params)
 
     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 +120,22 @@ 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,
-        )
+        build_params = None
+        if params is not None:
+            build_params = self._build_parameters(
+                params={
+                    "columns": params.columns,
+                    "sample_count": params.sample_count,
+                    "filters": params.filters,
+                    "sort_by": params.sort_by,
+                    "limit": params.limit,
+                    "offset": params.offset,
+                }
+            )
 
         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=build_params)
 
     def _get_execution_results_by_url(
         self, url: str, params: Optional[Dict[str, Any]] = None