Separate batch send from batch receive; cache request funcs:

- Separate the batch send and batch receive functions for persistent
  connection providers in order to deterministically cache the request
  information and be able to retrieve it without any request id guesswork.

- Remove the idea of a batching request counter / id 🎉

- Correct typing expectations for ``BatchRequestInformation``

Author: fselmo

tests/core/providers/test_async_http_provider.py

@@ -12,9 +12,6 @@
     AsyncWeb3,
     __version__ as web3py_version,
 )
-from web3._utils.batching import (
-    is_batching_context,
-)
 from web3.eth import (
     AsyncEth,
 )
@@ -131,4 +128,4 @@ async def test_async_http_empty_batch_response(mock_async_post):
         with pytest.raises(Web3RPCError, match="empty batch"):
             await batch.async_execute()
 
-    assert not is_batching_context()
+    assert not async_w3.provider._is_batching

tests/core/providers/test_async_ipc_provider.py

@@ -20,9 +20,6 @@
 from web3 import (
     AsyncWeb3,
 )
-from web3._utils.batching import (
-    is_batching_context,
-)
 from web3.datastructures import (
     AttributeDict,
 )
@@ -388,8 +385,8 @@ async def test_persistent_connection_provider_empty_batch_response(
                 )
             )
             async with async_w3.batch_requests() as batch:
-                assert is_batching_context()
+                assert async_w3.provider._is_batching
                 await batch.async_execute()
 
         # assert that even though there was an error, we have reset the batching state
-        assert not is_batching_context()
+        assert not async_w3.provider._is_batching

tests/core/providers/test_http_provider.py

@@ -15,9 +15,6 @@
     Web3,
     __version__ as web3py_version,
 )
-from web3._utils.batching import (
-    is_batching_context,
-)
 from web3.eth import (
     Eth,
 )
@@ -130,9 +127,9 @@ def test_http_empty_batch_response(mock_post):
     )
     w3 = Web3(HTTPProvider())
     with w3.batch_requests() as batch:
-        assert is_batching_context()
+        assert w3.provider._is_batching
         with pytest.raises(Web3RPCError, match="empty batch"):
             batch.execute()
 
     # assert that even though there was an error, we have reset the batching state
-    assert not is_batching_context()
+    assert not w3.provider._is_batching

tests/core/providers/test_websocket_provider.py

@@ -18,9 +18,6 @@
 from web3 import (
     AsyncWeb3,
 )
-from web3._utils.batching import (
-    is_batching_context,
-)
 from web3._utils.caching import (
     RequestInformation,
     generate_cache_key,
@@ -479,7 +476,7 @@ async def test_persistent_connection_provider_empty_batch_response():
         with pytest.raises(Web3RPCError, match="empty batch"):
             async with AsyncWeb3(WebSocketProvider("ws://mocked")) as async_w3:
                 async with async_w3.batch_requests() as batch:
-                    assert is_batching_context()
+                    assert async_w3.provider._is_batching
                     async_w3.provider._ws.recv = AsyncMock()
                     async_w3.provider._ws.recv.return_value = (
                         b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, '
@@ -489,7 +486,7 @@ async def test_persistent_connection_provider_empty_batch_response():
 
         # assert that even though there was an error, we have reset the batching
         # state
-        assert not is_batching_context()
+        assert not async_w3.provider._is_batching
 
 
 @pytest.mark.parametrize(

web3/_utils/batching.py

@@ -1,7 +1,3 @@
-import contextvars
-from copy import (
-    copy,
-)
 from types import (
     TracebackType,
 )
@@ -13,8 +9,6 @@
     Dict,
     Generic,
     List,
-    Optional,
-    Sequence,
     Tuple,
     Type,
     Union,
@@ -29,7 +23,6 @@
     Web3ValueError,
 )
 from web3.types import (
-    RPCEndpoint,
     TFunc,
     TReturn,
 )
@@ -58,24 +51,14 @@
         JSONBaseProvider,
     )
     from web3.types import (  # noqa: F401
+        RPCEndpoint,
         RPCResponse,
     )
 
 
 BATCH_REQUEST_ID = "batch_request"  # for use as the cache key for batch requests
 
-# control batching context via a context var
-_batching_context: contextvars.ContextVar[
-    Optional["RequestBatcher[Any]"]
-] = contextvars.ContextVar("batching_context", default=None)
-
-
-def is_batching_context() -> bool:
-    """Check if we're currently in a batching context."""
-    return _batching_context.get() is not None
-
-
-BatchRequestInformation = Tuple[Tuple["RPCEndpoint", Any], Sequence[Any]]
+BatchRequestInformation = Tuple[Tuple["RPCEndpoint", Any], Tuple[Any, ...]]
 RPC_METHODS_UNSUPPORTED_DURING_BATCH = {
     "eth_subscribe",
     "eth_unsubscribe",
@@ -105,22 +88,19 @@ def _provider(self) -> Union["JSONBaseProvider", "AsyncJSONBaseProvider"]:
         )
 
     def _validate_is_batching(self) -> None:
-        if not is_batching_context():
+        if not self._provider._is_batching:
             raise Web3ValueError(
                 "Batch has already been executed or cancelled. Create a new batch to "
                 "issue batched requests."
             )
 
     def _initialize_batching(self) -> None:
-        _batching_context.set(self)
+        self._provider._batching_context.set(self)
         self.clear()
 
     def _end_batching(self) -> None:
-        _batching_context.set(None)
         self.clear()
-        if self._provider.has_persistent_connection:
-            provider = cast("PersistentConnectionProvider", self._provider)
-            provider._batch_request_counter = None
+        self._provider._batching_context.set(None)
 
     def add(self, batch_payload: TReturn) -> None:
         self._validate_is_batching()
@@ -164,9 +144,6 @@ def execute(self) -> List["RPCResponse"]:
     def clear(self) -> None:
         self._requests_info = []
         self._async_requests_info = []
-        if self._provider.has_persistent_connection:
-            provider = cast("PersistentConnectionProvider", self._provider)
-            provider._batch_request_counter = next(copy(provider.request_counter))
 
     def cancel(self) -> None:
         self._end_batching()
@@ -189,9 +166,14 @@ def __exit__(
 
     async def async_execute(self) -> List["RPCResponse"]:
         self._validate_is_batching()
-        responses = await self.web3.manager._async_make_batch_request(
-            self._async_requests_info
-        )
+        if self._provider.has_persistent_connection:
+            responses = await self.web3.manager._async_make_socket_batch_request(
+                self._async_requests_info
+            )
+        else:
+            responses = await self.web3.manager._async_make_batch_request(
+                self._async_requests_info
+            )
         self._end_batching()
         return responses
 

web3/_utils/module_testing/web3_module.py

@@ -25,9 +25,6 @@
     AsyncWeb3,
     Web3,
 )
-from web3._utils.batching import (
-    is_batching_context,
-)
 from web3._utils.ens import (
     ens_addresses,
 )
@@ -345,7 +342,7 @@ def test_batch_requests(self, w3: "Web3", math_contract: Contract) -> None:
 
             # assert proper batch cleanup after execution
             assert batch._requests_info == []
-            assert not is_batching_context()
+            assert not w3.provider._is_batching
 
             # assert batch cannot be added to after execution
             with pytest.raises(
@@ -404,7 +401,7 @@ def test_batch_requests_initialized_as_object(
 
         # assert proper batch cleanup after execution
         assert batch._requests_info == []
-        assert not is_batching_context()
+        assert not w3.provider._is_batching
 
         # assert batch cannot be added to after execution
         with pytest.raises(
@@ -597,7 +594,7 @@ async def test_batch_requests(  # type: ignore[override]
 
             # assert proper batch cleanup after execution
             assert batch._async_requests_info == []
-            assert not is_batching_context()
+            assert not async_w3.provider._is_batching
 
             # assert batch cannot be added to after execution
             with pytest.raises(
@@ -660,7 +657,7 @@ async def test_batch_requests_initialized_as_object(  # type: ignore[override]
 
         # assert proper batch cleanup after execution
         assert batch._async_requests_info == []
-        assert not is_batching_context()
+        assert not async_w3.provider._is_batching
 
         # assert batch cannot be added to after execution
         with pytest.raises(
@@ -785,19 +782,20 @@ async def test_batch_requests_raises_for_common_unsupported_methods(  # type: ig
     async def test_batch_requests_concurrently_with_regular_requests(  # type: ignore[override]  # noqa: E501
         self, async_w3: AsyncWeb3  # type: ignore[override]
     ) -> None:
-        num_requests = 40
         responses = []
         batch_response = []
 
+        num_blocks = await async_w3.eth.block_number
+
         async def make_regular_requests() -> None:
-            for _ in range(num_requests):
-                responses.append(await async_w3.eth.get_block(0))
+            for i in range(num_blocks):
+                responses.append(await async_w3.eth.get_block(i))
                 await asyncio.sleep(0.01)
 
         async def make_batch_request() -> None:
             async with async_w3.batch_requests() as batch:
-                for _ in range(num_requests):
-                    batch.add(async_w3.eth.get_block(0))
+                for i in range(num_blocks):
+                    batch.add(async_w3.eth.get_block(i))
                     await asyncio.sleep(0.01)
                 batch_response.extend(await batch.async_execute())
 
@@ -806,7 +804,7 @@ async def make_batch_request() -> None:
             make_batch_request(),
         )
 
-        assert len(responses) == num_requests
-        assert len(batch_response) == num_requests
+        assert len(responses) == num_blocks
+        assert len(batch_response) == num_blocks
         assert all(SOME_BLOCK_KEYS.issubset(response.keys()) for response in responses)
         assert set(responses) == set(batch_response)

web3/contract/utils.py

@@ -44,7 +44,7 @@
     async_fill_transaction_defaults,
 )
 from web3._utils.batching import (
-    is_batching_context,
+    BatchRequestInformation,
 )
 from web3._utils.compat import (
     TypeAlias,
@@ -178,10 +178,8 @@ def call_contract_function(
     if abi_callable["type"] == "function":
         output_types = get_abi_output_types(abi_callable)
 
-    w3.provider
-    if is_batching_context():
-        BatchingReturnData: TypeAlias = Tuple[Tuple[RPCEndpoint, Any], Tuple[Any, ...]]
-        request_information = tuple(cast(BatchingReturnData, return_data))
+    if w3.provider._is_batching:
+        request_information = tuple(cast(BatchRequestInformation, return_data))
         method_and_params = request_information[0]
 
         # append return data formatting to result formatters
@@ -477,7 +475,7 @@ async def async_call_contract_function(
     if fn_abi["type"] == "function":
         output_types = get_abi_output_types(fn_abi)
 
-    if is_batching_context():
+    if async_w3.provider._is_batching:
         contract_call_return_data_formatter = format_contract_call_return_data_curried(
             async_w3,
             decode_tuples,
@@ -486,33 +484,24 @@ async def async_call_contract_function(
             normalizers,
             output_types,
         )
-        if async_w3.provider.has_persistent_connection:
-            # get the current request id
-            provider = cast("PersistentConnectionProvider", async_w3.provider)
-            current_request_id = provider._batch_request_counter - 1
-            provider._request_processor.append_result_formatter_for_request(
-                current_request_id, contract_call_return_data_formatter
-            )
-        else:
-            BatchingReturnData: TypeAlias = Tuple[
-                Tuple[RPCEndpoint, Any], Tuple[Any, ...]
-            ]
-            request_information = tuple(cast(BatchingReturnData, return_data))
-            method_and_params = request_information[0]
-
-            # append return data formatter to result formatters
-            current_response_formatters = request_information[1]
-            current_result_formatters = current_response_formatters[0]
-            updated_result_formatters = compose(
-                contract_call_return_data_formatter,
-                current_result_formatters,
-            )
-            response_formatters = (
-                updated_result_formatters,  # result formatters
-                current_response_formatters[1],  # error formatters
-                current_response_formatters[2],  # null result formatters
-            )
-            return (method_and_params, response_formatters)
+
+        BatchingReturnData: TypeAlias = Tuple[Tuple[RPCEndpoint, Any], Tuple[Any, ...]]
+        request_information = tuple(cast(BatchingReturnData, return_data))
+        method_and_params = request_information[0]
+
+        # append return data formatter to result formatters
+        current_response_formatters = request_information[1]
+        current_result_formatters = current_response_formatters[0]
+        updated_result_formatters = compose(
+            contract_call_return_data_formatter,
+            current_result_formatters,
+        )
+        response_formatters = (
+            updated_result_formatters,  # result formatters
+            current_response_formatters[1],  # error formatters
+            current_response_formatters[2],  # null result formatters
+        )
+        return (method_and_params, response_formatters)
 
         return return_data