chore: refactor out _apply_request_formatters

Author: BobTheBuidler

tests/core/method-class/test_method.py

@@ -17,7 +17,6 @@
 )
 from web3.method import (
     Method,
-    _apply_request_formatters,
     default_root_munger,
 )
 from web3.module import (
@@ -61,11 +60,7 @@ def test_get_formatters_default_formatter_for_falsy_config():
     default_result_formatters = method.result_formatters(
         method.method_selector_fn(), "some module"
     )
-    assert _apply_request_formatters(["a", "b", "c"], default_request_formatters) == (
-        "a",
-        "b",
-        "c",
-    )
+    assert default_request_formatters(["a", "b", "c"]) == ("a", "b", "c")
     assert apply_result_formatters(default_result_formatters, ["a", "b", "c"]) == [
         "a",
         "b",

web3/_utils/method_formatters.py

@@ -1069,7 +1069,7 @@ def get_request_formatters(
         PYTHONIC_REQUEST_FORMATTERS,
     )
     formatters = combine_formatters(request_formatter_maps, method_name)
-    return compose(*formatters)
+    return compose(tuple, *formatters)
 
 
 def raise_block_not_found(params: Tuple[BlockIdentifier, bool]) -> NoReturn:

web3/manager.py

@@ -6,7 +6,6 @@
     AsyncGenerator,
     Callable,
     Coroutine,
-    Dict,
     List,
     Optional,
     Sequence,
@@ -48,6 +47,7 @@
 )
 from web3.method import (
     Method,
+    ResponseFormatters,
 )
 from web3.middleware import (
     AttributeDictMiddleware,
@@ -360,9 +360,7 @@ async def socket_request(
         self,
         method: RPCEndpoint,
         params: Any,
-        response_formatters: Optional[
-            Tuple[Dict[str, Callable[..., Any]], Callable[..., Any], Callable[..., Any]]
-        ] = None,
+        response_formatters: Optional[ResponseFormatters[Any]] = None,
     ) -> RPCResponse:
         provider = cast(PersistentConnectionProvider, self._provider)
         self.logger.debug(

web3/method.py

@@ -3,24 +3,15 @@
     TYPE_CHECKING,
     Any,
     Callable,
-    Dict,
     Generic,
     List,
     Optional,
     Sequence,
     Tuple,
     Type,
-    Union,
 )
 import warnings
 
-from eth_utils.curried import (
-    to_tuple,
-)
-from eth_utils.toolz import (
-    pipe,
-)
-
 from web3._utils.batching import (
     RPC_METHODS_UNSUPPORTED_DURING_BATCH,
 )
@@ -40,6 +31,8 @@
     Web3ValueError,
 )
 from web3.types import (
+    RequestFormatter,
+    ResponseFormatter,
     RPCEndpoint,
     TFunc,
     TReturn,
@@ -54,16 +47,10 @@
 
 
 Munger = Callable[..., Any]
-
-
-@to_tuple
-def _apply_request_formatters(
-    params: Any, request_formatters: Dict[RPCEndpoint, Callable[..., TReturn]]
-) -> Tuple[Any, ...]:
-    if request_formatters:
-        formatted_params = pipe(params, request_formatters)
-        return formatted_params
-    return params
+RequestArgs = Tuple[RPCEndpoint, Sequence[Any]]
+ResponseFormatters = Tuple[
+    ResponseFormatter[TReturn], ResponseFormatter[TReturn], ResponseFormatter[TReturn]
+]
 
 
 def _set_mungers(
@@ -136,9 +123,15 @@ def __init__(
         self,
         json_rpc_method: Optional[RPCEndpoint] = None,
         mungers: Optional[Sequence[Munger]] = None,
-        request_formatters: Optional[Callable[..., TReturn]] = None,
-        result_formatters: Optional[Callable[..., TReturn]] = None,
-        null_result_formatters: Optional[Callable[..., TReturn]] = None,
+        request_formatters: Optional[
+            Callable[[RPCEndpoint], RequestFormatter[Any]]
+        ] = None,
+        result_formatters: Optional[
+            Callable[[RPCEndpoint, "Module"], ResponseFormatter[TReturn]]
+        ] = None,
+        null_result_formatters: Optional[
+            Callable[[RPCEndpoint], ResponseFormatter[TReturn]]
+        ] = None,
         method_choice_depends_on_args: Optional[Callable[..., RPCEndpoint]] = None,
         is_property: bool = False,
     ):
@@ -202,14 +195,7 @@ def input_munger(self, module: "Module", args: Any, kwargs: Any) -> List[Any]:
 
     def process_params(
         self, module: "Module", *args: Any, **kwargs: Any
-    ) -> Tuple[
-        Tuple[Union[RPCEndpoint, Callable[..., RPCEndpoint]], Tuple[RPCEndpoint, ...]],
-        Tuple[
-            Union[TReturn, Dict[str, Callable[..., Any]]],
-            Callable[..., Any],
-            Union[TReturn, Callable[..., Any]],
-        ],
-    ]:
+    ) -> Tuple[RequestArgs, ResponseFormatters[TReturn]]:
         params = self.input_munger(module, args, kwargs)
 
         if self.method_choice_depends_on_args:
@@ -232,10 +218,12 @@ def process_params(
             get_error_formatters(method),
             self.null_result_formatters(method),
         )
-        request = (
-            method,
-            _apply_request_formatters(params, self.request_formatters(method)),
-        )
+
+        if request_formatters := self.request_formatters(method):
+            params = request_formatters(params)
+
+        request = method, params
+
         return request, response_formatters
 
 

web3/module.py

@@ -5,11 +5,10 @@
     Coroutine,
     Dict,
     Optional,
-    Sequence,
     Tuple,
     TypeVar,
     Union,
-    cast,
+    overload,
 )
 
 from eth_abi.codec import (
@@ -27,14 +26,16 @@
 )
 from web3.method import (
     Method,
+    RequestArgs,
+    ResponseFormatters,
 )
 from web3.providers.persistent import (
     PersistentConnectionProvider,
 )
 from web3.types import (
     FormattedEthSubscriptionResponse,
-    RPCEndpoint,
     RPCResponse,
+    TFunc,
 )
 
 if TYPE_CHECKING:
@@ -58,34 +59,52 @@ def apply_result_formatters(
 TReturn = TypeVar("TReturn")
 
 
+@overload
+def retrieve_request_information_for_batching(
+    w3: "AsyncWeb3",
+    module: "Module",
+    method: Method[Callable[..., Any]],
+) -> Callable[..., Coroutine[Any, Any, Tuple[RequestArgs, ResponseFormatters[Any]]]]:
+    ...
+
+
+@overload
+def retrieve_request_information_for_batching(
+    w3: "Web3",
+    module: "Module",
+    method: Method[Callable[..., Any]],
+) -> Callable[..., Tuple[RequestArgs, ResponseFormatters[Any]]]:
+    ...
+
+
 @curry
 def retrieve_request_information_for_batching(
     w3: Union["AsyncWeb3", "Web3"],
     module: "Module",
     method: Method[Callable[..., Any]],
 ) -> Union[
-    Callable[..., Tuple[Tuple[RPCEndpoint, Any], Sequence[Any]]],
-    Callable[..., Coroutine[Any, Any, Tuple[Tuple[RPCEndpoint, Any], Sequence[Any]]]],
+    Callable[..., Tuple[RequestArgs, ResponseFormatters[Any]]],
+    Callable[..., Coroutine[Any, Any, Tuple[RequestArgs, ResponseFormatters[Any]]]],
 ]:
     async def async_inner(
         *args: Any, **kwargs: Any
-    ) -> Tuple[Tuple[RPCEndpoint, Any], Sequence[Any]]:
+    ) -> Tuple[RequestArgs, ResponseFormatters[Any]]:
+        response_formatters: ResponseFormatters[Any]
         (method_str, params), response_formatters = method.process_params(
             module, *args, **kwargs
         )
         if isinstance(w3.provider, PersistentConnectionProvider):
             w3.provider._request_processor.cache_request_information(
-                None, cast(RPCEndpoint, method_str), params, response_formatters
+                None, method_str, params, response_formatters
             )
-        return (cast(RPCEndpoint, method_str), params), response_formatters
+        return (method_str, params), response_formatters
 
-    def inner(
-        *args: Any, **kwargs: Any
-    ) -> Tuple[Tuple[RPCEndpoint, Any], Sequence[Any]]:
+    def inner(*args: Any, **kwargs: Any) -> Tuple[RequestArgs, ResponseFormatters[Any]]:
+        response_formatters: ResponseFormatters[Any]
         (method_str, params), response_formatters = method.process_params(
             module, *args, **kwargs
         )
-        return (cast(RPCEndpoint, method_str), params), response_formatters
+        return (method_str, params), response_formatters
 
     return async_inner if module.is_async else inner
 
@@ -97,6 +116,8 @@ def retrieve_blocking_method_call_fn(
     method: Method[Callable[..., TReturn]],
 ) -> Callable[..., Union[TReturn, LogFilter]]:
     def caller(*args: Any, **kwargs: Any) -> Union[TReturn, LogFilter]:
+        response_formatters: ResponseFormatters[Any]
+
         try:
             (method_str, params), response_formatters = method.process_params(
                 module, *args, **kwargs
@@ -133,6 +154,8 @@ def retrieve_async_method_call_fn(
     async def caller(
         *args: Any, **kwargs: Any
     ) -> Union[RPCResponse, FormattedEthSubscriptionResponse, AsyncLogFilter]:
+        response_formatters: ResponseFormatters[Any]
+
         try:
             (method_str, params), response_formatters = method.process_params(
                 module, *args, **kwargs
@@ -142,7 +165,7 @@ async def caller(
 
         if isinstance(async_w3.provider, PersistentConnectionProvider):
             return await async_w3.manager.socket_request(
-                cast(RPCEndpoint, method_str),
+                method_str,
                 params,
                 response_formatters=response_formatters,
             )
@@ -167,12 +190,22 @@ async def caller(
 class Module:
     is_async = False
 
+    retrieve_request_information: Callable[
+        [Method[TFunc]],
+        Union[
+            Callable[..., Tuple[RequestArgs, ResponseFormatters[Any]]],
+            Callable[
+                ..., Coroutine[Any, Any, Tuple[RequestArgs, ResponseFormatters[Any]]]
+            ],
+        ],
+    ]
+
     def __init__(self, w3: Union["AsyncWeb3", "Web3"]) -> None:
         if self.is_async:
             self.retrieve_caller_fn = retrieve_async_method_call_fn(w3, self)
         else:
             self.retrieve_caller_fn = retrieve_blocking_method_call_fn(w3, self)
-        self.retrieve_request_information = retrieve_request_information_for_batching(
+        self.retrieve_request_information = retrieve_request_information_for_batching(  # type: ignore [call-overload]
             w3, self
         )
         self.w3 = w3