asyncify get_storage_at

docs/providers.rst

@@ -429,6 +429,7 @@ Eth
 - :meth:`web3.eth.get_transaction() <web3.eth.Eth.get_transaction>`
 - :meth:`web3.eth.get_transaction_count() <web3.eth.Eth.get_transaction_count>`
 - :meth:`web3.eth.get_transaction_receipt() <web3.eth.Eth.get_transaction_receipt>`
+- :meth:`web3.eth.get_storage_at() <web3.eth.Eth.get_storage_at>`
 - :meth:`web3.eth.send_transaction() <web3.eth.Eth.send_transaction>`
 - :meth:`web3.eth.send_raw_transaction() <web3.eth.Eth.send_raw_transaction>`
 - :meth:`web3.eth.wait_for_transaction_receipt() <web3.eth.Eth.wait_for_transaction_receipt>`

newsfragments/2350.feature.rst

@@ -0,0 +1 @@
++ Add async `eth.get_storage_at` method

web3/_utils/module_testing/eth_module.py

@@ -1096,6 +1096,30 @@ async def test_async_eth_syncing(self, async_w3: "Web3") -> None:
             assert is_integer(sync_dict['currentBlock'])
             assert is_integer(sync_dict['highestBlock'])
 
+    @pytest.mark.asyncio
+    async def test_async_eth_get_storage_at(
+        self, async_w3: "Web3", emitter_contract_address: ChecksumAddress
+    ) -> None:
+        storage = await async_w3.eth.get_storage_at(emitter_contract_address, 0)  # type: ignore
+        assert isinstance(storage, HexBytes)
+
+    @pytest.mark.asyncio
+    @pytest.mark.xfail
+    async def test_async_eth_get_storage_at_ens_name(
+        self, async_w3: "Web3", emitter_contract_address: ChecksumAddress
+    ) -> None:
+        with ens_addresses(async_w3, {'emitter.eth': emitter_contract_address}):
+            storage = await async_w3.eth.get_storage_at('emitter.eth', 0)  # type: ignore
+            assert isinstance(storage, HexBytes)
+
+    @pytest.mark.asyncio
+    async def test_async_eth_get_storage_at_invalid_address(self, async_w3: "Web3") -> None:
+        coinbase = await async_w3.eth.coinbase  # type: ignore
+        with pytest.raises(InvalidAddress):
+            await async_w3.eth.get_storage_at(
+                ChecksumAddress(HexAddress(HexStr(coinbase.lower()))),
+                0)  # type: ignore
+
     def test_async_provider_default_account(
         self,
         async_w3: "Web3",

web3/eth.py

@@ -283,6 +283,16 @@ def block_id_munger(
             block_identifier = self.default_block
         return (account, block_identifier)
 
+    def get_storage_at_munger(
+        self,
+        account: Union[Address, ChecksumAddress, ENS],
+        position: int,
+        block_identifier: Optional[BlockIdentifier] = None
+    ) -> Tuple[Union[Address, ChecksumAddress, ENS], int, BlockIdentifier]:
+        if block_identifier is None:
+            block_identifier = self.default_block
+        return (account, position, block_identifier)
+
     def call_munger(
         self,
         transaction: TxParams,
@@ -519,6 +529,19 @@ async def _wait_for_tx_receipt_with_timeout(
                 f"after {timeout} seconds"
             )
 
+    _get_storage_at: Method[Callable[..., Awaitable[HexBytes]]] = Method(
+        RPC.eth_getStorageAt,
+        mungers=[BaseEth.get_storage_at_munger],
+    )
+
+    async def get_storage_at(
+        self,
+        account: Union[Address, ChecksumAddress, ENS],
+        position: int,
+        block_identifier: Optional[BlockIdentifier] = None
+    ) -> HexBytes:
+        return await self._get_storage_at(account, position, block_identifier)
+
     async def call(
         self,
         transaction: TxParams,
@@ -636,19 +659,9 @@ def max_priority_fee(self) -> Wei:
             )
             return fee_history_priority_fee(self)
 
-    def get_storage_at_munger(
-        self,
-        account: Union[Address, ChecksumAddress, ENS],
-        position: int,
-        block_identifier: Optional[BlockIdentifier] = None
-    ) -> Tuple[Union[Address, ChecksumAddress, ENS], int, BlockIdentifier]:
-        if block_identifier is None:
-            block_identifier = self.default_block
-        return (account, position, block_identifier)
-
     get_storage_at: Method[Callable[..., HexBytes]] = Method(
         RPC.eth_getStorageAt,
-        mungers=[get_storage_at_munger],
+        mungers=[BaseEth.get_storage_at_munger],
     )
 
     def get_proof_munger(