Repository: ethereum/web3.py

Issue:
Improvements for `*ContractEvent` and `*ContractFunction` APIs
### What feature should we add?

I see some low-hanging fruit in contract interactions with functions and events.

- `contract.myFunction.signature` & `contract.MyEvent.signature`
  - We could use a clearer attribute to signify `signature`. Right now this is the same thing as `abi_element_identifier`, though I'm not sure off the top of my head if this always returns the signature.
  
- `contract.MyEvent.to_topic()` (`.as_topic()`, `.topic`?)
  - We could use an API that returns the event as a topic (`w3.keccak(text=event_signature).to_0x_hex()`). This would significantly improve subscribing to topics on contracts:
  
  ```py
  w3.eth.subscribe(
      "logs", 
      {
          "address": my_contract.address,
          "topics": [
              my_contract.events.MyEvent.to_topic()
          ]
      }
  )
  ```



Discussion:

Patch:
diff --git a/docs/web3.contract.rst b/docs/web3.contract.rst
index fa1c6e86..d3ab7490 100644
--- a/docs/web3.contract.rst
+++ b/docs/web3.contract.rst
@@ -1108,43 +1108,47 @@ For example:
         myContract = web3.eth.contract(address=contract_address, abi=contract_abi)
         tx_hash = myContract.functions.myFunction().transact()
         receipt = web3.eth.get_transaction_receipt(tx_hash)
-        myContract.events.myEvent().process_receipt(receipt)
+        myContract.events.MyEvent().process_receipt(receipt)
 
 .. py:class:: ContractEvent
 
 Attributes
 ~~~~~~~~~~
 
-The :py:class:`ContractEvent` class provides attributes for each event. Access the event attributes through `Contract.events.myEvent`.
+The :py:class:`ContractEvent` class provides attributes for each event. Access the event attributes through `Contract.events.MyEvent`.
 
-.. py:attribute:: ContractEvent.myEvent(*args, **kwargs).abi_element_identifier
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).abi_element_identifier
 
     The signature of the event assigned to the class ``__name__`` during initialization.
 
-.. py:attribute:: ContractEvent.myEvent(*args, **kwargs).name
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).name
 
     A string representing the event, receive or fallback name.
 
-    Use :py:attr:`ContractEvent.myEvent(*args, **kwargs).signature` when the event arguments are needed.
+    Use :py:attr:`ContractEvent.MyEvent(*args, **kwargs).signature` when the event arguments are needed.
 
-    This is an alias of :py:attr:`ContractEvent.myEvent(*args, **kwargs).event_name`.
+    This is an alias of :py:attr:`ContractEvent.MyEvent(*args, **kwargs).event_name`.
 
-.. py:attribute:: ContractEvent.myEvent(*args, **kwargs).signature
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).signature
 
     A string representing the event signature.
 
-.. py:attribute:: ContractEvent.myEvent(*args, **kwargs).abi
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).abi
 
     The event ABI with the type, name, inputs.
 
-.. py:attribute:: ContractEvent.myEvent(*args, **kwargs).argument_names
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).argument_names
 
     The event input names.
 
-.. py:attribute:: ContractEvent.myEvent(*args, **kwargs).argument_types
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).argument_types
 
     The event input types.
 
+.. py:attribute:: ContractEvent.MyEvent(*args, **kwargs).topic
+
+    The event topic represented by a hex encoded string from the keccak signature.
+
 Methods
 ~~~~~~~
 
@@ -1172,9 +1176,9 @@ Methods
 
         my_contract = web3.eth.contract(address=contract_address, abi=contract_abi)
 
-        # get ``myEvent`` logs from block 1337 to block 2337 where the value for the
+        # get ``MyEvent`` logs from block 1337 to block 2337 where the value for the
         # event argument "eventArg1" is either 1, 2, or 3
-        my_contract.events.myEvent().get_logs(
+        my_contract.events.MyEvent().get_logs(
             argument_filters={"eventArg1": [1, 2, 3]},
             from_block=1337,
             to_block=2337,
@@ -1187,14 +1191,14 @@ Methods
 
    Extracts the pertinent logs from a transaction receipt.
 
-   If there are no errors, ``process_receipt`` returns a tuple of :ref:`Event Log Objects <event-log-object>`, emitted from the event (e.g. ``myEvent``),
+   If there are no errors, ``process_receipt`` returns a tuple of :ref:`Event Log Objects <event-log-object>`, emitted from the event (e.g. ``MyEvent``),
    with decoded output.
 
    .. code-block:: python
 
        >>> tx_hash = contract.functions.myFunction(12345).transact({'to':contract_address})
        >>> tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
-       >>> rich_logs = contract.events.myEvent().process_receipt(tx_receipt)
+       >>> rich_logs = contract.events.MyEvent().process_receipt(tx_receipt)
        >>> rich_logs[0]['args']
        {'myArg': 12345}
 
@@ -1211,12 +1215,12 @@ Methods
 
        >>> tx_hash = contract.functions.myFunction(12345).transact({'to':contract_address})
        >>> tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
-       >>> processed_logs = contract.events.myEvent().process_receipt(tx_receipt)
+       >>> processed_logs = contract.events.MyEvent().process_receipt(tx_receipt)
        >>> processed_logs
        (
           AttributeDict({
               'args': AttributeDict({}),
-              'event': 'myEvent',
+              'event': 'MyEvent',
               'logIndex': 0,
               'transactionIndex': 0,
               'transactionHash': HexBytes('0xfb95ccb6ab39e19821fb339dee33e7afe2545527725b61c64490a5613f8d11fa'),
@@ -1229,7 +1233,7 @@ Methods
 
        # Or, if there were errors encountered during processing:
        >>> from web3.logs import STRICT, IGNORE, DISCARD, WARN
-       >>> processed_logs = contract.events.myEvent().process_receipt(tx_receipt, errors=IGNORE)
+       >>> processed_logs = contract.events.MyEvent().process_receipt(tx_receipt, errors=IGNORE)
        >>> processed_logs
        (
            AttributeDict({
@@ -1247,7 +1251,7 @@ Methods
                'errors': LogTopicError('Expected 1 log topics.  Got 0')})
           })
        )
-       >>> processed_logs = contract.events.myEvent().process_receipt(tx_receipt, errors=DISCARD)
+       >>> processed_logs = contract.events.MyEvent().process_receipt(tx_receipt, errors=DISCARD)
        >>> assert processed_logs == ()
        True
 
@@ -1261,11 +1265,11 @@ Methods
        >>> tx_hash = contract.functions.myFunction(12345).transact({'to':contract_address})
        >>> tx_receipt = w3.eth.get_transaction_receipt(tx_hash)
        >>> log_to_process = tx_receipt['logs'][0]
-       >>> processed_log = contract.events.myEvent().process_log(log_to_process)
+       >>> processed_log = contract.events.MyEvent().process_log(log_to_process)
        >>> processed_log
        AttributeDict({
            'args': AttributeDict({}),
-           'event': 'myEvent',
+           'event': 'MyEvent',
            'logIndex': 0,
            'transactionIndex': 0,
            'transactionHash': HexBytes('0xfb95ccb6ab39e19821fb339dee33e7afe2545527725b61c64490a5613f8d11fa'),
diff --git a/newsfragments/3585.bugfix.rst b/newsfragments/3585.bugfix.rst
new file mode 100644
index 00000000..2c3babd2
--- /dev/null
+++ b/newsfragments/3585.bugfix.rst
@@ -0,0 +1,1 @@
+Handle the case when a single RPC error response is returned for a batch request, instead of always expecting a list of responses.
diff --git a/newsfragments/3586.docs.rst b/newsfragments/3586.docs.rst
new file mode 100644
index 00000000..65f7f919
--- /dev/null
+++ b/newsfragments/3586.docs.rst
@@ -0,0 +1,1 @@
+Include ``topic`` attribute in ``ContractEvent`` docs.
diff --git a/web3/manager.py b/web3/manager.py
index 8e2f2b64..2a3e6d11 100644
--- a/web3/manager.py
+++ b/web3/manager.py
@@ -8,6 +8,7 @@ from typing import (
     Coroutine,
     Dict,
     List,
+    NoReturn,
     Optional,
     Sequence,
     Tuple,
@@ -266,6 +267,30 @@ def _validate_response(
         _raise_bad_response_format(response)
 
 
+def _raise_error_for_batch_response(
+    response: RPCResponse,
+    logger: Optional[logging.Logger] = None,
+) -> NoReturn:
+    error = response.get("error")
+    if error is None:
+        _raise_bad_response_format(
+            response,
+            "Batch response must be formatted as a list of responses or "
+            "as a single JSON-RPC error response.",
+        )
+    _validate_response(
+        response,
+        None,
+        is_subscription_response=False,
+        logger=logger,
+        params=[],
+    )
+    # This should not be reached, but if it is, raise a generic `BadResponseFormat`
+    raise BadResponseFormat(
+        "Batch response was in an unexpected format and unable to be parsed."
+    )
+
+
 class RequestManager:
     logger = logging.getLogger("web3.manager.RequestManager")
 
@@ -436,17 +461,23 @@ class RequestManager:
         request_func = provider.batch_request_func(
             cast("Web3", self.w3), cast("MiddlewareOnion", self.middleware_onion)
         )
-        responses = request_func(
+        response = request_func(
             [
                 (method, params)
                 for (method, params), _response_formatters in requests_info
             ]
         )
-        formatted_responses = [
-            self._format_batched_response(info, resp)
-            for info, resp in zip(requests_info, responses)
-        ]
-        return list(formatted_responses)
+
+        if isinstance(response, list):
+            # expected format
+            formatted_responses = [
+                self._format_batched_response(info, cast(RPCResponse, resp))
+                for info, resp in zip(requests_info, response)
+            ]
+            return list(formatted_responses)
+        else:
+            # expect a single response with an error
+            _raise_error_for_batch_response(response, self.logger)
 
     async def _async_make_batch_request(
         self,
@@ -465,25 +496,31 @@ class RequestManager:
         # since we add items to the batch without awaiting, we unpack the coroutines
         # and await them all here
         unpacked_requests_info = await asyncio.gather(*requests_info)
-        responses = await request_func(
+        response = await request_func(
             [
                 (method, params)
                 for (method, params), _response_formatters in unpacked_requests_info
             ]
         )
 
-        if isinstance(self.provider, PersistentConnectionProvider):
-            # call _process_response for each response in the batch
-            return [
-                cast(RPCResponse, await self._process_response(resp))
-                for resp in responses
+        if isinstance(response, list):
+            # expected format
+            response = cast(List[RPCResponse], response)
+            if isinstance(self.provider, PersistentConnectionProvider):
+                # call _process_response for each response in the batch
+                return [
+                    cast(RPCResponse, await self._process_response(resp))
+                    for resp in response
+                ]
+
+            formatted_responses = [
+                self._format_batched_response(info, resp)
+                for info, resp in zip(unpacked_requests_info, response)
             ]
-
-        formatted_responses = [
-            self._format_batched_response(info, resp)
-            for info, resp in zip(unpacked_requests_info, responses)
-        ]
-        return list(formatted_responses)
+            return list(formatted_responses)
+        else:
+            # expect a single response with an error
+            _raise_error_for_batch_response(response, self.logger)
 
     def _format_batched_response(
         self,
@@ -491,6 +528,13 @@ class RequestManager:
         response: RPCResponse,
     ) -> RPCResponse:
         result_formatters, error_formatters, null_result_formatters = requests_info[1]
+        _validate_response(
+            response,
+            error_formatters,
+            is_subscription_response=False,
+            logger=self.logger,
+            params=requests_info[0][1],
+        )
         return apply_result_formatters(
             result_formatters,
             self.formatted_response(
diff --git a/web3/middleware/base.py b/web3/middleware/base.py
index 590fce58..5f968f82 100644
--- a/web3/middleware/base.py
+++ b/web3/middleware/base.py
@@ -62,15 +62,19 @@ class Web3Middleware:
     ) -> "MakeBatchRequestFn":
         def middleware(
             requests_info: List[Tuple["RPCEndpoint", Any]]
-        ) -> List["RPCResponse"]:
+        ) -> Union[List["RPCResponse"], "RPCResponse"]:
             req_processed = [
                 self.request_processor(method, params)
                 for (method, params) in requests_info
             ]
-            responses = make_batch_request(req_processed)
+            response = make_batch_request(req_processed)
+            if not isinstance(response, list):
+                # RPC errors return only one response with the error object
+                return response
+
             methods, _params = zip(*req_processed)
             formatted_responses = [
-                self.response_processor(m, r) for m, r in zip(methods, responses)
+                self.response_processor(m, r) for m, r in zip(methods, response)
             ]
             return formatted_responses
 
@@ -103,16 +107,20 @@ class Web3Middleware:
     ) -> "AsyncMakeBatchRequestFn":
         async def middleware(
             requests_info: List[Tuple["RPCEndpoint", Any]]
-        ) -> List["RPCResponse"]:
+        ) -> Union[List["RPCResponse"], "RPCResponse"]:
             req_processed = [
                 await self.async_request_processor(method, params)
                 for (method, params) in requests_info
             ]
-            responses = await make_batch_request(req_processed)
+            response = await make_batch_request(req_processed)
+            if not isinstance(response, list):
+                # RPC errors return only one response with the error object
+                return response
+
             methods, _params = zip(*req_processed)
             formatted_responses = [
                 await self.async_response_processor(m, r)
-                for m, r in zip(methods, responses)
+                for m, r in zip(methods, response)
             ]
             return formatted_responses
 
diff --git a/web3/providers/async_base.py b/web3/providers/async_base.py
index 55a4a8ee..56c4fe71 100644
--- a/web3/providers/async_base.py
+++ b/web3/providers/async_base.py
@@ -77,7 +77,8 @@ class AsyncBaseProvider:
 
     _is_batching: bool = False
     _batch_request_func_cache: Tuple[
-        Tuple[Middleware, ...], Callable[..., Coroutine[Any, Any, List[RPCResponse]]]
+        Tuple[Middleware, ...],
+        Callable[..., Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]]],
     ] = (None, None)
 
     is_async = True
@@ -119,7 +120,7 @@ class AsyncBaseProvider:
 
     async def batch_request_func(
         self, async_w3: "AsyncWeb3", middleware_onion: MiddlewareOnion
-    ) -> Callable[..., Coroutine[Any, Any, List[RPCResponse]]]:
+    ) -> Callable[..., Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]]]:
         middleware: Tuple[Middleware, ...] = middleware_onion.as_tuple_of_middleware()
 
         cache_key = self._batch_request_func_cache[0]
@@ -141,8 +142,8 @@ class AsyncBaseProvider:
 
     async def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
-        raise NotImplementedError("Only AsyncHTTPProvider supports this method")
+    ) -> Union[List[RPCResponse], RPCResponse]:
+        raise NotImplementedError("Providers must implement this method")
 
     async def is_connected(self, show_traceback: bool = False) -> bool:
         raise NotImplementedError("Providers must implement this method")
diff --git a/web3/providers/base.py b/web3/providers/base.py
index e1606967..36a1ca9f 100644
--- a/web3/providers/base.py
+++ b/web3/providers/base.py
@@ -118,7 +118,7 @@ class JSONBaseProvider(BaseProvider):
 
     _is_batching: bool = False
     _batch_request_func_cache: Tuple[
-        Tuple[Middleware, ...], Callable[..., List[RPCResponse]]
+        Tuple[Middleware, ...], Callable[..., Union[List[RPCResponse], RPCResponse]]
     ] = (None, None)
 
     def __init__(self, **kwargs: Any) -> None:
@@ -168,7 +168,7 @@ class JSONBaseProvider(BaseProvider):
 
     def batch_request_func(
         self, w3: "Web3", middleware_onion: MiddlewareOnion
-    ) -> Callable[..., List[RPCResponse]]:
+    ) -> Callable[..., Union[List[RPCResponse], RPCResponse]]:
         middleware: Tuple[Middleware, ...] = middleware_onion.as_tuple_of_middleware()
 
         cache_key = self._batch_request_func_cache[0]
@@ -199,5 +199,5 @@ class JSONBaseProvider(BaseProvider):
 
     def make_batch_request(
         self, requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         raise NotImplementedError("Providers must implement this method")
diff --git a/web3/providers/persistent/request_processor.py b/web3/providers/persistent/request_processor.py
index 9d5bf30a..65d4d998 100644
--- a/web3/providers/persistent/request_processor.py
+++ b/web3/providers/persistent/request_processor.py
@@ -6,6 +6,7 @@ from typing import (
     Callable,
     Dict,
     Generic,
+    List,
     Optional,
     Tuple,
     TypeVar,
@@ -270,6 +271,15 @@ class RequestProcessor:
 
     # raw response cache
 
+    def _is_batch_response(
+        self, raw_response: Union[List[RPCResponse], RPCResponse]
+    ) -> bool:
+        return isinstance(raw_response, list) or (
+            isinstance(raw_response, dict)
+            and raw_response.get("id") is None
+            and self._provider._is_batching
+        )
+
     async def cache_raw_response(
         self, raw_response: Any, subscription: bool = False
     ) -> None:
@@ -296,7 +306,7 @@ class RequestProcessor:
                 # otherwise, put it in the subscription response queue so a response
                 # can be yielded by the message stream
                 await self._subscription_response_queue.put(raw_response)
-        elif isinstance(raw_response, list):
+        elif self._is_batch_response(raw_response):
             # Since only one batch should be in the cache at all times, we use a
             # constant cache key for the batch response.
             cache_key = generate_cache_key(BATCH_REQUEST_ID)
diff --git a/web3/providers/rpc/async_rpc.py b/web3/providers/rpc/async_rpc.py
index bc8ef17d..85e27c33 100644
--- a/web3/providers/rpc/async_rpc.py
+++ b/web3/providers/rpc/async_rpc.py
@@ -168,15 +168,20 @@ class AsyncHTTPProvider(AsyncJSONBaseProvider):
 
     async def make_batch_request(
         self, batch_requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         self.logger.debug(f"Making batch request HTTP - uri: `{self.endpoint_uri}`")
         request_data = self.encode_batch_rpc_request(batch_requests)
         raw_response = await self._request_session_manager.async_make_post_request(
             self.endpoint_uri, request_data, **self.get_request_kwargs()
         )
         self.logger.debug("Received batch response HTTP.")
-        responses_list = cast(List[RPCResponse], self.decode_rpc_response(raw_response))
-        return sort_batch_response_by_response_ids(responses_list)
+        response = self.decode_rpc_response(raw_response)
+        if not isinstance(response, list):
+            # RPC errors return only one response with the error object
+            return response
+        return sort_batch_response_by_response_ids(
+            cast(List[RPCResponse], sort_batch_response_by_response_ids(response))
+        )
 
     async def disconnect(self) -> None:
         cache = self._request_session_manager.session_cache
diff --git a/web3/providers/rpc/rpc.py b/web3/providers/rpc/rpc.py
index 7dde0416..4c41d46e 100644
--- a/web3/providers/rpc/rpc.py
+++ b/web3/providers/rpc/rpc.py
@@ -176,12 +176,17 @@ class HTTPProvider(JSONBaseProvider):
 
     def make_batch_request(
         self, batch_requests: List[Tuple[RPCEndpoint, Any]]
-    ) -> List[RPCResponse]:
+    ) -> Union[List[RPCResponse], RPCResponse]:
         self.logger.debug(f"Making batch request HTTP, uri: `{self.endpoint_uri}`")
         request_data = self.encode_batch_rpc_request(batch_requests)
         raw_response = self._request_session_manager.make_post_request(
             self.endpoint_uri, request_data, **self.get_request_kwargs()
         )
         self.logger.debug("Received batch response HTTP.")
-        responses_list = cast(List[RPCResponse], self.decode_rpc_response(raw_response))
-        return sort_batch_response_by_response_ids(responses_list)
+        response = self.decode_rpc_response(raw_response)
+        if not isinstance(response, list):
+            # RPC errors return only one response with the error object
+            return response
+        return sort_batch_response_by_response_ids(
+            cast(List[RPCResponse], sort_batch_response_by_response_ids(response))
+        )
diff --git a/web3/types.py b/web3/types.py
index f7d0acc7..5a4fe72b 100644
--- a/web3/types.py
+++ b/web3/types.py
@@ -301,10 +301,13 @@ class CreateAccessListResponse(TypedDict):
 
 
 MakeRequestFn = Callable[[RPCEndpoint, Any], RPCResponse]
-MakeBatchRequestFn = Callable[[List[Tuple[RPCEndpoint, Any]]], List[RPCResponse]]
+MakeBatchRequestFn = Callable[
+    [List[Tuple[RPCEndpoint, Any]]], Union[List[RPCResponse], RPCResponse]
+]
 AsyncMakeRequestFn = Callable[[RPCEndpoint, Any], Coroutine[Any, Any, RPCResponse]]
 AsyncMakeBatchRequestFn = Callable[
-    [List[Tuple[RPCEndpoint, Any]]], Coroutine[Any, Any, List[RPCResponse]]
+    [List[Tuple[RPCEndpoint, Any]]],
+    Coroutine[Any, Any, Union[List[RPCResponse], RPCResponse]],
 ]
 
 


Test patch:
diff --git a/tests/core/providers/test_async_http_provider.py b/tests/core/providers/test_async_http_provider.py
index 558d260b..4ae96224 100644
--- a/tests/core/providers/test_async_http_provider.py
+++ b/tests/core/providers/test_async_http_provider.py
@@ -1,4 +1,8 @@
 import pytest
+from unittest.mock import (
+    AsyncMock,
+    patch,
+)
 
 from aiohttp import (
     ClientSession,
@@ -13,6 +17,7 @@ from web3.eth import (
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    Web3RPCError,
 )
 from web3.geth import (
     AsyncGeth,
@@ -107,3 +112,21 @@ def test_get_request_headers(provider):
         headers["User-Agent"] == f"web3.py/{web3py_version}/"
         f"{AsyncHTTPProvider.__module__}.{AsyncHTTPProvider.__qualname__}"
     )
+
+
+@patch(
+    "web3._utils.http_session_manager.HTTPSessionManager.async_make_post_request",
+    new_callable=AsyncMock,
+)
+@pytest.mark.asyncio
+async def test_async_http_empty_batch_response(mock_async_post):
+    mock_async_post.return_value = (
+        b'{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}'
+    )
+    async_w3 = AsyncWeb3(AsyncHTTPProvider())
+    async with async_w3.batch_requests() as batch:
+        with pytest.raises(Web3RPCError, match="empty batch"):
+            await batch.async_execute()
+
+    # assert that even though there was an error, we have reset the batching state
+    assert not async_w3.provider._is_batching
diff --git a/tests/core/providers/test_async_ipc_provider.py b/tests/core/providers/test_async_ipc_provider.py
index c559a445..132d3a1a 100644
--- a/tests/core/providers/test_async_ipc_provider.py
+++ b/tests/core/providers/test_async_ipc_provider.py
@@ -25,6 +25,7 @@ from web3.datastructures import (
 )
 from web3.exceptions import (
     ReadBufferLimitReached,
+    Web3RPCError,
 )
 from web3.providers import (
     AsyncIPCProvider,
@@ -366,3 +367,24 @@ async def test_async_ipc_provider_write_messages_end_with_new_line_delimiter(
 
         request_data = b'{"id": 0, "jsonrpc": "2.0", "method": "method", "params": []}'
         w3.provider._writer.write.assert_called_with(request_data + b"\n")
+
+
+@pytest.mark.asyncio
+async def test_persistent_connection_provider_empty_batch_response(
+    simple_ipc_server,
+    jsonrpc_ipc_pipe_path,
+):
+    async with AsyncWeb3(
+        AsyncIPCProvider(pathlib.Path(jsonrpc_ipc_pipe_path))
+    ) as async_w3:
+        async_w3.provider._reader.readline = AsyncMock()
+        async_w3.provider._reader.readline.return_value = (
+            b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, "message": '
+            b'"empty batch"}}\n'
+        )
+        async with async_w3.batch_requests() as batch:
+            with pytest.raises(Web3RPCError, match="empty batch"):
+                await batch.async_execute()
+
+        # assert that even though there was an error, we have reset the batching state
+        assert not async_w3.provider._is_batching
diff --git a/tests/core/providers/test_http_provider.py b/tests/core/providers/test_http_provider.py
index 82a982de..49b1b2ba 100644
--- a/tests/core/providers/test_http_provider.py
+++ b/tests/core/providers/test_http_provider.py
@@ -1,4 +1,8 @@
 import pytest
+from unittest.mock import (
+    Mock,
+    patch,
+)
 
 from requests import (
     Session,
@@ -16,6 +20,7 @@ from web3.eth import (
 )
 from web3.exceptions import (
     ProviderConnectionError,
+    Web3RPCError,
 )
 from web3.geth import (
     Geth,
@@ -110,3 +115,20 @@ def test_get_request_headers(provider):
         headers["User-Agent"] == f"web3.py/{web3py_version}/"
         f"{HTTPProvider.__module__}.{HTTPProvider.__qualname__}"
     )
+
+
+@patch(
+    "web3._utils.http_session_manager.HTTPSessionManager.make_post_request",
+    new_callable=Mock,
+)
+def test_http_empty_batch_response(mock_post):
+    mock_post.return_value = (
+        b'{"jsonrpc":"2.0","id":null,"error":{"code":-32600,"message":"empty batch"}}'
+    )
+    w3 = Web3(HTTPProvider())
+    with w3.batch_requests() as batch:
+        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 w3.provider._is_batching
diff --git a/tests/core/providers/test_websocket_provider.py b/tests/core/providers/test_websocket_provider.py
index 7df7fbd8..c3205670 100644
--- a/tests/core/providers/test_websocket_provider.py
+++ b/tests/core/providers/test_websocket_provider.py
@@ -27,6 +27,7 @@ from web3._utils.module_testing.module_testing_utils import (
 )
 from web3.exceptions import (
     TimeExhausted,
+    Web3RPCError,
 )
 from web3.providers.persistent import (
     WebSocketProvider,
@@ -432,3 +433,24 @@ async def test_listener_task_breaks_out_of_handle_subscriptions_when_cancelled()
     # up the exception.
     with pytest.raises(asyncio.CancelledError):
         await async_w3.subscription_manager.handle_subscriptions(run_forever=True)
+
+
+@pytest.mark.asyncio
+async def test_persistent_connection_provider_empty_batch_response():
+    with patch(
+        "web3.providers.persistent.websocket.connect",
+        new=lambda *_1, **_2: _mocked_ws_conn(),
+    ):
+        async with AsyncWeb3(WebSocketProvider("ws://mocked")) as async_w3:
+            async_w3.provider._ws.recv = AsyncMock()
+            async_w3.provider._ws.recv.return_value = (
+                b'{"jsonrpc": "2.0","id":null,"error": {"code": -32600, "message": '
+                b'"empty batch"}}\n'
+            )
+            async with async_w3.batch_requests() as batch:
+                with pytest.raises(Web3RPCError, match="empty batch"):
+                    await batch.async_execute()
+
+            # assert that even though there was an error, we have reset the batching
+            # state
+            assert not async_w3.provider._is_batching