diff --git a/async_substrate_interface/async_substrate.py b/async_substrate_interface/async_substrate.py index c827377..b7441bb 100644 --- a/async_substrate_interface/async_substrate.py +++ b/async_substrate_interface/async_substrate.py @@ -784,9 +784,9 @@ async def retrieve(self, item_id: str) -> Optional[dict]: if item is not None: if item.done(): self.max_subscriptions.release() + res = item.result() del self._received[item_id] - - return item.result() + return res else: try: return self._received_subscriptions[item_id].get_nowait() @@ -1165,7 +1165,7 @@ async def get_runtime_for_version( async def _get_runtime_for_version( self, runtime_version: int, block_hash: Optional[str] = None ) -> Runtime: - runtime_config = RuntimeConfigurationObject() + runtime_config = RuntimeConfigurationObject(ss58_format=self.ss58_format) runtime_config.clear_type_registry() runtime_config.update_type_registry(load_type_registry_preset(name="core")) @@ -2337,7 +2337,7 @@ async def _make_rpc_request( request_manager.add_request(item_id, payload["id"]) while True: - for item_id in list(request_manager.response_map.keys()): + for item_id in request_manager.unresponded(): if ( item_id not in request_manager.responses or asyncio.iscoroutinefunction(result_handler) @@ -2368,7 +2368,6 @@ async def _make_rpc_request( runtime=runtime, force_legacy_decode=force_legacy_decode, ) - request_manager.add_response( item_id, decoded_response, complete ) diff --git a/async_substrate_interface/sync_substrate.py b/async_substrate_interface/sync_substrate.py index 44fd158..692db3b 100644 --- a/async_substrate_interface/sync_substrate.py +++ b/async_substrate_interface/sync_substrate.py @@ -1924,7 +1924,7 @@ def _make_rpc_request( _received[response["params"]["subscription"]] = response else: raise SubstrateRequestException(response) - for item_id in list(request_manager.response_map.keys()): + for item_id in request_manager.unresponded(): if item_id not in request_manager.responses or isinstance( result_handler, Callable ): diff --git a/async_substrate_interface/types.py b/async_substrate_interface/types.py index f1efbc3..ecacca8 100644 --- a/async_substrate_interface/types.py +++ b/async_substrate_interface/types.py @@ -418,6 +418,15 @@ def get_results(self) -> RequestResults: request_id: info["results"] for request_id, info in self.responses.items() } + def unresponded(self): + """ + Yields items from response_map whose corresponding response is missing or incomplete. + """ + for item_id, request_id in list(self.response_map.items()): + response_info = self.responses.get(request_id) + if response_info is None or not response_info["complete"]: + yield item_id + @dataclass class Preprocessed: diff --git a/tests/integration_tests/test_async_substrate_interface.py b/tests/integration_tests/test_async_substrate_interface.py index 2c50213..098111d 100644 --- a/tests/integration_tests/test_async_substrate_interface.py +++ b/tests/integration_tests/test_async_substrate_interface.py @@ -126,9 +126,26 @@ async def test_get_events_proper_decoding(): async with AsyncSubstrateInterface(ARCHIVE_ENTRYPOINT) as substrate: all_events = await substrate.get_events(block_hash=block_hash) event = all_events[1] - print(type(event["attributes"])) assert event["attributes"] == ( "5G1NjW9YhXLadMWajvTkfcJy6up3yH2q1YzMXDTi6ijanChe", 30, "0xa6b4e5c8241d60ece0c25056b19f7d21ae845269fc771ad46bf3e011865129a5", ) + + +@pytest.mark.asyncio +async def test_query_multiple(): + block = 6153277 + cks = [ + "5FH9AQM4kqbkdC9jyV5FrdEWVYt41nkhFstop7Vhyfb9ZsXt", + "5GQxLKxjZWNZDsghmYcw7P6ahC7XJCjx1WD94WGh92quSycx", + "5EcaPiDT1cv951SkCFsvdHDs2yAEUWhJDuRP9mHb343WnaVn", + ] + async with AsyncSubstrateInterface(ARCHIVE_ENTRYPOINT) as substrate: + block_hash = await substrate.get_block_hash(block_id=block) + assert await substrate.query_multiple( + params=cks, + module="SubtensorModule", + storage_function="OwnedHotkeys", + block_hash=block_hash, + ) diff --git a/tests/integration_tests/test_substrate_interface.py b/tests/integration_tests/test_substrate_interface.py index be4eb29..27e134a 100644 --- a/tests/integration_tests/test_substrate_interface.py +++ b/tests/integration_tests/test_substrate_interface.py @@ -78,9 +78,25 @@ def test_get_events_proper_decoding(): with SubstrateInterface(ARCHIVE_ENTRYPOINT) as substrate: all_events = substrate.get_events(block_hash=block_hash) event = all_events[1] - print(type(event["attributes"])) assert event["attributes"] == ( "5G1NjW9YhXLadMWajvTkfcJy6up3yH2q1YzMXDTi6ijanChe", 30, "0xa6b4e5c8241d60ece0c25056b19f7d21ae845269fc771ad46bf3e011865129a5", ) + + +def test_query_multiple(): + block = 6153277 + cks = [ + "5FH9AQM4kqbkdC9jyV5FrdEWVYt41nkhFstop7Vhyfb9ZsXt", + "5GQxLKxjZWNZDsghmYcw7P6ahC7XJCjx1WD94WGh92quSycx", + "5EcaPiDT1cv951SkCFsvdHDs2yAEUWhJDuRP9mHb343WnaVn", + ] + with SubstrateInterface(ARCHIVE_ENTRYPOINT) as substrate: + block_hash = substrate.get_block_hash(block_id=block) + assert substrate.query_multiple( + params=cks, + module="SubtensorModule", + storage_function="OwnedHotkeys", + block_hash=block_hash, + ) diff --git a/tests/unit_tests/asyncio_/test_substrate_interface.py b/tests/unit_tests/asyncio_/test_substrate_interface.py index 817cdf3..d4d692f 100644 --- a/tests/unit_tests/asyncio_/test_substrate_interface.py +++ b/tests/unit_tests/asyncio_/test_substrate_interface.py @@ -3,6 +3,7 @@ import pytest from websockets.exceptions import InvalidURI +from websockets.protocol import State from async_substrate_interface.async_substrate import AsyncSubstrateInterface from async_substrate_interface.types import ScaleObj @@ -103,9 +104,9 @@ async def test_websocket_shutdown_timer(): async with AsyncSubstrateInterface("wss://lite.sub.latent.to:443") as substrate: await substrate.get_chain_head() await asyncio.sleep(6) - assert ( - substrate.ws._initialized is False - ) # connection should have closed automatically + assert ( + substrate.ws.state is State.CLOSED + ) # connection should have closed automatically # using custom ws shutdown timer of 10.0 seconds async with AsyncSubstrateInterface( @@ -113,7 +114,7 @@ async def test_websocket_shutdown_timer(): ) as substrate: await substrate.get_chain_head() await asyncio.sleep(6) # same sleep time as before - assert substrate.ws._initialized is True # connection should still be open + assert substrate.ws.state is State.OPEN # connection should still be open @pytest.mark.asyncio