More performant attrdict implementation:

Speeds up the attrdict middleware by nearly 2x.

5000 calls with a large block static response, removing any latency
from the server:
  - ~3.5 seconds w/ no middleware
  - ~7.2 seconds w/ attrdict middleware on this branch (3.7s more)
  - ~10.3 seconds w/ attrdict middleware on main (6.8s more)

6.8/3.7 = 1.84x

Changes:

- Simplify the recursive method of ``AttributeDict`` by removing the
  dependence on ``recursive_map`` and using a simpler recursive approach
  to nested lists and dicts.

- Don't use ``assoc`` as this ends up creating a new dict every time,
  adding more overhead.

Author: fselmo

newsfragments/3575.performance.rst

@@ -0,0 +1 @@
+Improve ``AttributeDict.recursive()`` and ``AttributeDictMiddleware`` performance, effectively speeding up response processing for attrdict middleware by nearly 2x.

web3/datastructures.py

@@ -15,7 +15,6 @@
     Optional,
     Sequence,
     Tuple,
-    Type,
     TypeVar,
     Union,
     cast,
@@ -25,9 +24,6 @@
     is_integer,
 )
 
-from web3._utils.formatters import (
-    recursive_map,
-)
 from web3.exceptions import (
     Web3AssertionError,
     Web3TypeError,
@@ -81,19 +77,18 @@ def _repr_pretty_(self, builder: Any, cycle: bool) -> None:
         builder.text(")")
 
     @classmethod
-    def _apply_if_mapping(cls: Type[T], value: TValue) -> Union[T, TValue]:
+    def recursive(cls, value: TValue) -> Any:
+        """
+        Recursively convert mappings to ReadableAttributeDict instances and
+        process nested collections (e.g., lists, sets, and dictionaries).
+        """
         if isinstance(value, Mapping):
-            # error: Too many arguments for "object"
-            return cls(value)  # type: ignore
-        else:
-            return value
-
-    @classmethod
-    def recursive(cls, value: TValue) -> "ReadableAttributeDict[TKey, TValue]":
-        return cast(
-            "ReadableAttributeDict[TKey, TValue]",
-            recursive_map(cls._apply_if_mapping, value),
-        )
+            return cls({k: cls.recursive(v) for k, v in value.items()})
+        elif isinstance(value, Sequence) and not isinstance(value, (str, bytes)):
+            return type(value)([cls.recursive(v) for v in value])  # type: ignore
+        elif isinstance(value, set):
+            return {cls.recursive(v) for v in value}
+        return value
 
 
 class MutableAttributeDict(

web3/middleware/attrdict.py

@@ -7,10 +7,6 @@
     cast,
 )
 
-from eth_utils.toolz import (
-    assoc,
-)
-
 from web3.datastructures import (
     AttributeDict,
 )
@@ -33,21 +29,18 @@
 
 
 def _handle_async_response(response: "RPCResponse") -> "RPCResponse":
+    """
+    Process the RPC response by converting nested dictionaries into AttributeDict.
+    """
     if "result" in response:
-        return assoc(response, "result", AttributeDict.recursive(response["result"]))
+        response["result"] = AttributeDict.recursive(response["result"])
     elif "params" in response and "result" in response["params"]:
-        # this is a subscription response
-        return assoc(
-            response,
-            "params",
-            assoc(
-                response["params"],
-                "result",
-                AttributeDict.recursive(response["params"]["result"]),
-            ),
+        # subscription response
+        response["params"]["result"] = AttributeDict.recursive(
+            response["params"]["result"]
         )
-    else:
-        return response
+
+    return response
 
 
 class AttributeDictMiddleware(Web3Middleware, ABC):
@@ -60,19 +53,16 @@ class AttributeDictMiddleware(Web3Middleware, ABC):
 
     def response_processor(self, method: "RPCEndpoint", response: "RPCResponse") -> Any:
         if "result" in response:
-            return assoc(
-                response, "result", AttributeDict.recursive(response["result"])
-            )
-        else:
-            return response
+            new_result = AttributeDict.recursive(response["result"])
+            response = {**response, "result": new_result}
+        return response
 
     # -- async -- #
 
     async def async_response_processor(
         self, method: "RPCEndpoint", response: "RPCResponse"
     ) -> Any:
         if self._w3.provider.has_persistent_connection:
-            # asynchronous response processing
             provider = cast("PersistentConnectionProvider", self._w3.provider)
             provider._request_processor.append_middleware_response_processor(
                 response, _handle_async_response