Merge pull request #272 from opentensor/release/1.6.2

Release/1.6.2

Author: ibraheem-abe

.github/workflows/e2e-tests.yml

@@ -124,7 +124,7 @@ jobs:
         os:
           - ubuntu-latest
         test-file: ${{ fromJson(needs.find-tests.outputs.test-files) }}
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
     steps:
       - name: Check-out repository
         uses: actions/checkout@v4

.github/workflows/unit-and-integration-test.yml

@@ -15,7 +15,7 @@ jobs:
       fail-fast: false
       max-parallel: 5
       matrix:
-        python-version: ["3.9", "3.10", "3.11", "3.12", "3.13", "3.14"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 
     steps:
       - name: Checkout repository

CHANGELOG.md

@@ -1,5 +1,17 @@
 # Changelog
 
+## 1.6.2 /2025-02-19
+
+## What's Changed
+* Typing by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/265
+* Cache Improvements by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/267
+* improve (async) query map result by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/266
+* Use threaded bt-decode by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/268
+* Feat: Handle new error message fmt for InBlock source by @ibraheem-abe in https://github.com/opentensor/async-substrate-interface/pull/269
+* Update: Remove python 3.9 support by @ibraheem-abe in https://github.com/opentensor/async-substrate-interface/pull/271
+
+**Full Changelog**: https://github.com/opentensor/async-substrate-interface/compare/v1.6.1...v1.6.2
+
 ## 1.6.1 /2025-02-03
 * RuntimeCache updates by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/260
 * fix memory leak by @thewhaleking in https://github.com/opentensor/async-substrate-interface/pull/261

async_substrate_interface/async_substrate.py

@@ -31,7 +31,6 @@
 from scalecodec.types import (
     GenericCall,
     GenericExtrinsic,
-    GenericRuntimeCallDefinition,
     ss58_encode,
     MultiAccountId,
 )
@@ -74,12 +73,10 @@
     _bt_decode_to_dict_or_list,
     legacy_scale_decode,
     convert_account_ids,
+    decode_query_map_async,
 )
 from async_substrate_interface.utils.storage import StorageKey
 from async_substrate_interface.type_registry import _TYPE_REGISTRY
-from async_substrate_interface.utils.decoding import (
-    decode_query_map,
-)
 
 ResultHandler = Callable[[dict, Any], Awaitable[tuple[dict, bool]]]
 
@@ -526,6 +523,18 @@ async def retrieve_next_page(self, start_key) -> list:
         self.last_key = result.last_key
         return result.records
 
+    async def retrieve_all_records(self) -> list[Any]:
+        """
+        Retrieves all records from all subsequent pages for the AsyncQueryMapResult,
+        returning them as a list.
+
+        Side effect:
+            The self.records list will be populated fully after running this method.
+        """
+        async for _ in self:
+            pass
+        return self.records
+
     def __aiter__(self):
         return self
 
@@ -558,6 +567,7 @@ async def __anext__(self):
             self.loading_complete = True
             raise StopAsyncIteration
 
+        self.records.extend(next_page)
         # Update the buffer with the newly fetched records
         self._buffer = iter(next_page)
         return next(self._buffer)
@@ -1408,7 +1418,9 @@ async def decode_scale(
             if runtime is None:
                 runtime = await self.init_runtime(block_hash=block_hash)
             if runtime.metadata_v15 is not None and force_legacy is False:
-                obj = decode_by_type_string(type_string, runtime.registry, scale_bytes)
+                obj = await asyncio.to_thread(
+                    decode_by_type_string, type_string, runtime.registry, scale_bytes
+                )
                 if self.decode_ss58:
                     try:
                         type_str_int = int(type_string.split("::")[1])
@@ -2762,19 +2774,34 @@ async def rpc_request(
             logger.error(f"Substrate Request Exception: {result[payload_id]}")
             raise SubstrateRequestException(result[payload_id][0])
 
-    @cached_fetcher(max_size=SUBSTRATE_CACHE_METHOD_SIZE)
-    async def get_block_hash(self, block_id: int) -> str:
+    async def get_block_hash(self, block_id: Optional[int]) -> str:
         """
-        Retrieves the hash of the specified block number
+        Retrieves the hash of the specified block number, or the chaintip if None
         Args:
             block_id: block number
 
         Returns:
             Hash of the block
         """
+        if block_id is None:
+            return await self.get_chain_head()
+        else:
+            if (block_hash := self.runtime_cache.blocks.get(block_id)) is not None:
+                return block_hash
+
+            block_hash = await self._cached_get_block_hash(block_id)
+            self.runtime_cache.add_item(block_hash=block_hash, block=block_id)
+            return block_hash
+
+    @cached_fetcher(max_size=SUBSTRATE_CACHE_METHOD_SIZE)
+    async def _cached_get_block_hash(self, block_id: int) -> str:
+        """
+        The design of this method is as such, because it allows for an easy drop-in for a different cache, such
+        as is the case with DiskCachedAsyncSubstrateInterface._cached_get_block_hash
+        """
         return await self._get_block_hash(block_id)
 
-    async def _get_block_hash(self, block_id: int) -> str:
+    async def _get_block_hash(self, block_id: Optional[int]) -> str:
         return (await self.rpc_request("chain_getBlockHash", [block_id]))["result"]
 
     async def get_chain_head(self) -> str:
@@ -3852,18 +3879,20 @@ async def query_map(
                     params=[result_keys, block_hash],
                     runtime=runtime,
                 )
+                changes = []
                 for result_group in response["result"]:
-                    result = decode_query_map(
-                        result_group["changes"],
-                        prefix,
-                        runtime,
-                        param_types,
-                        params,
-                        value_type,
-                        key_hashers,
-                        ignore_decoding_errors,
-                        self.decode_ss58,
-                    )
+                    changes.extend(result_group["changes"])
+                result = await decode_query_map_async(
+                    changes,
+                    prefix,
+                    runtime,
+                    param_types,
+                    params,
+                    value_type,
+                    key_hashers,
+                    ignore_decoding_errors,
+                    self.decode_ss58,
+                )
             else:
                 # storage item and value scale type are not included here because this is batch-decoded in rust
                 page_batches = [
@@ -3881,8 +3910,8 @@ async def query_map(
                 results: RequestResults = await self._make_rpc_request(
                     payloads, runtime=runtime
                 )
-                for result in results.values():
-                    res = result[0]
+                for result_ in results.values():
+                    res = result_[0]
                     if "error" in res:
                         err_msg = res["error"]["message"]
                         if (
@@ -3900,7 +3929,7 @@ async def query_map(
                     else:
                         for result_group in res["result"]:
                             changes.extend(result_group["changes"])
-                result = decode_query_map(
+                result = await decode_query_map_async(
                     changes,
                     prefix,
                     runtime,
@@ -4113,6 +4142,14 @@ async def result_handler(message: dict, subscription_id) -> tuple[dict, bool]:
                         "extrinsic_hash": "0x{}".format(extrinsic.extrinsic_hash.hex()),
                         "finalized": False,
                     }, True
+
+            elif "params" in message and message["params"].get("result") == "invalid":
+                failure_message = f"Subscription {subscription_id} invalid: {message}"
+                async with self.ws as ws:
+                    await ws.unsubscribe(subscription_id)
+                logger.error(failure_message)
+                raise SubstrateRequestException(failure_message)
+
             return message, False
 
         if wait_for_inclusion or wait_for_finalization:
@@ -4250,13 +4287,25 @@ async def get_metadata_event(
 
     async def get_block_number(self, block_hash: Optional[str] = None) -> int:
         """Async version of `substrateinterface.base.get_block_number` method."""
-        response = await self.rpc_request("chain_getHeader", [block_hash])
+        if block_hash is None:
+            return await self._get_block_number(None)
+        if (block := self.runtime_cache.blocks_reverse.get(block_hash)) is not None:
+            return block
+        block = await self._cached_get_block_number(block_hash)
+        self.runtime_cache.add_item(block_hash=block_hash, block=block)
+        return block
 
-        if response["result"]:
-            return int(response["result"]["number"], 16)
-        raise SubstrateRequestException(
-            f"Unable to retrieve block number for {block_hash}"
-        )
+    @cached_fetcher(max_size=SUBSTRATE_CACHE_METHOD_SIZE)
+    async def _cached_get_block_number(self, block_hash: str) -> int:
+        """
+        The design of this method is as such, because it allows for an easy drop-in for a different cache, such
+        as is the case with DiskCachedAsyncSubstrateInterface._cached_get_block_number
+        """
+        return await self._get_block_number(block_hash=block_hash)
+
+    async def _get_block_number(self, block_hash: Optional[str]) -> int:
+        response = await self.rpc_request("chain_getHeader", [block_hash])
+        return int(response["result"]["number"], 16)
 
     async def close(self):
         """
@@ -4351,9 +4400,13 @@ async def get_block_runtime_version_for(self, block_hash: str):
         return await self._get_block_runtime_version_for(block_hash)
 
     @async_sql_lru_cache(maxsize=SUBSTRATE_CACHE_METHOD_SIZE)
-    async def get_block_hash(self, block_id: int) -> str:
+    async def _cached_get_block_hash(self, block_id: int) -> str:
         return await self._get_block_hash(block_id)
 
+    @async_sql_lru_cache(maxsize=SUBSTRATE_CACHE_METHOD_SIZE)
+    async def _cached_get_block_number(self, block_hash: str) -> int:
+        return await self._get_block_number(block_hash=block_hash)
+
 
 async def get_async_substrate_interface(
     url: str,

async_substrate_interface/py.typed

@@ -482,6 +482,18 @@ def retrieve_next_page(self, start_key) -> list:
         self.last_key = result.last_key
         return result.records
 
+    def retrieve_all_records(self) -> list[Any]:
+        """
+        Retrieves all records from all subsequent pages for the QueryMapResult,
+        returning them as a list.
+
+        Side effect:
+            The self.records list will be populated fully after running this method.
+        """
+        for _ in self:
+            pass
+        return self.records
+
     def __iter__(self):
         return self
 
@@ -511,6 +523,7 @@ def __next__(self):
             self.loading_complete = True
             raise StopIteration
 
+        self.records.extend(next_page)
         # Update the buffer with the newly fetched records
         self._buffer = iter(next_page)
         return next(self._buffer)
@@ -2052,8 +2065,21 @@ def rpc_request(
         else:
             raise SubstrateRequestException(result[payload_id][0])
 
+    def get_block_hash(self, block_id: Optional[int]) -> str:
+        """
+        Retrieves the block hash for a given block number, or the chaintip hash if None
+        """
+        if block_id is None:
+            return self.get_chain_head()
+        else:
+            if (block_hash := self.runtime_cache.blocks.get(block_id)) is not None:
+                return block_hash
+            block_hash = self._get_block_hash(block_id)
+            self.runtime_cache.add_item(block_hash=block_hash, block=block_id)
+            return block_hash
+
     @functools.lru_cache(maxsize=SUBSTRATE_CACHE_METHOD_SIZE)
-    def get_block_hash(self, block_id: int) -> str:
+    def _get_block_hash(self, block_id: int) -> str:
         return self.rpc_request("chain_getBlockHash", [block_id])["result"]
 
     def get_chain_head(self) -> str:
@@ -3247,6 +3273,13 @@ def result_handler(message: dict, subscription_id) -> tuple[dict, bool]:
                         "extrinsic_hash": "0x{}".format(extrinsic.extrinsic_hash.hex()),
                         "finalized": False,
                     }, True
+
+            elif "params" in message and message["params"].get("result") == "invalid":
+                failure_message = f"Subscription {subscription_id} invalid: {message}"
+                self.rpc_request("author_unwatchExtrinsic", [subscription_id])
+                logger.error(failure_message)
+                raise SubstrateRequestException(failure_message)
+
             return message, False
 
         if wait_for_inclusion or wait_for_finalization:
@@ -3380,15 +3413,27 @@ def get_metadata_event(
         return self._get_metadata_event(module_name, event_name, runtime)
 
     def get_block_number(self, block_hash: Optional[str] = None) -> int:
-        """Async version of `substrateinterface.base.get_block_number` method."""
-        response = self.rpc_request("chain_getHeader", [block_hash])
-
-        if response["result"]:
-            return int(response["result"]["number"], 16)
+        """
+        Retrieves the block number for a given block hash or chaintip.
+        """
+        if block_hash is None:
+            return self._get_block_number(None)
         else:
-            raise SubstrateRequestException(
-                f"Unable to determine block number for {block_hash}"
-            )
+            if (
+                block_number := self.runtime_cache.blocks_reverse.get(block_hash)
+            ) is not None:
+                return block_number
+            block_number = self._cached_get_block_number(block_hash=block_hash)
+            self.runtime_cache.add_item(block_hash=block_hash, block=block_number)
+            return block_number
+
+    @functools.lru_cache(maxsize=SUBSTRATE_CACHE_METHOD_SIZE)
+    def _cached_get_block_number(self, block_hash: Optional[str]) -> int:
+        return self._get_block_number(block_hash=block_hash)
+
+    def _get_block_number(self, block_hash: Optional[str]) -> int:
+        response = self.rpc_request("chain_getHeader", [block_hash])
+        return int(response["result"]["number"], 16)
 
     def close(self):
         """
@@ -3404,6 +3449,7 @@ def close(self):
         self.get_block_runtime_info.cache_clear()
         self.get_block_runtime_version_for.cache_clear()
         self.supports_rpc_method.cache_clear()
-        self.get_block_hash.cache_clear()
+        self._get_block_hash.cache_clear()
+        self._cached_get_block_number.cache_clear()
 
     encode_scale = SubstrateMixin._encode_scale