Scrapped the multiprocessing idea. Now using decode_list from bt-decode

Author: thewhaleking

async_substrate_interface/async_substrate.py

@@ -56,14 +56,12 @@
 )
 from async_substrate_interface.utils.storage import StorageKey
 from async_substrate_interface.type_registry import _TYPE_REGISTRY
-from async_substrate_interface.utils.decoding_attempt import (
+from async_substrate_interface.utils.decoding import (
     decode_query_map,
-    _decode_scale_with_runtime,
 )
 
 if TYPE_CHECKING:
     from websockets.asyncio.client import ClientConnection
-    from concurrent.futures import ProcessPoolExecutor
 
 ResultHandler = Callable[[dict, Any], Awaitable[tuple[dict, bool]]]
 
@@ -418,7 +416,6 @@ def __init__(
         last_key: Optional[str] = None,
         max_results: Optional[int] = None,
         ignore_decoding_errors: bool = False,
-        executor: Optional["ProcessPoolExecutor"] = None,
     ):
         self.records = records
         self.page_size = page_size
@@ -431,7 +428,6 @@ def __init__(
         self.params = params
         self.ignore_decoding_errors = ignore_decoding_errors
         self.loading_complete = False
-        self.executor = executor
         self._buffer = iter(self.records)  # Initialize the buffer with initial records
 
     async def retrieve_next_page(self, start_key) -> list:
@@ -444,7 +440,6 @@ async def retrieve_next_page(self, start_key) -> list:
             start_key=start_key,
             max_results=self.max_results,
             ignore_decoding_errors=self.ignore_decoding_errors,
-            executor=self.executor,
         )
         if len(result.records) < self.page_size:
             self.loading_complete = True
@@ -2867,7 +2862,6 @@ async def query_map(
         page_size: int = 100,
         ignore_decoding_errors: bool = False,
         reuse_block_hash: bool = False,
-        executor: Optional["ProcessPoolExecutor"] = None,
     ) -> AsyncQueryMapResult:
         """
         Iterates over all key-pairs located at the given module and storage_function. The storage
@@ -2972,54 +2966,16 @@ async def query_map(
             if "error" in response:
                 raise SubstrateRequestException(response["error"]["message"])
             for result_group in response["result"]:
-                if executor:
-                    # print(
-                    #     ("prefix", type("prefix")),
-                    #     ("runtime_registry", type(runtime.registry)),
-                    #     ("param_types", type(param_types)),
-                    #     ("params", type(params)),
-                    #     ("value_type", type(value_type)),
-                    #     ("key_hasher", type(key_hashers)),
-                    #     ("ignore_decoding_errors", type(ignore_decoding_errors)),
-                    # )
-                    result = await asyncio.get_running_loop().run_in_executor(
-                        executor,
-                        decode_query_map,
-                        result_group["changes"],
-                        prefix,
-                        runtime.registry.registry,
-                        param_types,
-                        params,
-                        value_type,
-                        key_hashers,
-                        ignore_decoding_errors,
-                    )
-                    # max_workers = executor._max_workers
-                    # result_group_changes_groups = [result_group["changes"][i:i + max_workers] for i in range(0, len(result_group["changes"]), max_workers)]
-                    # all_results = executor.map(
-                    #     self._decode_query_map,
-                    #     result_group["changes"],
-                    #     repeat(prefix),
-                    #     repeat(runtime.registry),
-                    #     repeat(param_types),
-                    #     repeat(params),
-                    #     repeat(value_type),
-                    #     repeat(key_hashers),
-                    #     repeat(ignore_decoding_errors)
-                    # )
-                    # for r in all_results:
-                    #     result.extend(r)
-                else:
-                    result = decode_query_map(
-                        result_group["changes"],
-                        prefix,
-                        runtime.registry.registry,
-                        param_types,
-                        params,
-                        value_type,
-                        key_hashers,
-                        ignore_decoding_errors,
-                    )
+                result = decode_query_map(
+                    result_group["changes"],
+                    prefix,
+                    runtime,
+                    param_types,
+                    params,
+                    value_type,
+                    key_hashers,
+                    ignore_decoding_errors,
+                )
         return AsyncQueryMapResult(
             records=result,
             page_size=page_size,
@@ -3031,7 +2987,6 @@ async def query_map(
             last_key=last_key,
             max_results=max_results,
             ignore_decoding_errors=ignore_decoding_errors,
-            executor=executor,
         )
 
     async def submit_extrinsic(

async_substrate_interface/utils/decoding.py

@@ -1,6 +1,14 @@
-from typing import Union
+from typing import Union, TYPE_CHECKING
 
-from bt_decode import AxonInfo, PrometheusInfo
+from bt_decode import AxonInfo, PrometheusInfo, decode_list
+from scalecodec import ss58_encode
+from bittensor_wallet.utils import SS58_FORMAT
+
+from async_substrate_interface.utils import hex_to_bytes
+from async_substrate_interface.types import ScaleObj
+
+if TYPE_CHECKING:
+    from async_substrate_interface.types import Runtime
 
 
 def _determine_if_old_runtime_call(runtime_call_def, metadata_v15_value) -> bool:
@@ -44,3 +52,91 @@ def _bt_decode_to_dict_or_list(obj) -> Union[dict, list[dict]]:
             else:
                 as_dict[key] = val
     return as_dict
+
+
+def _decode_scale_list_with_runtime(
+    type_string: list[str],
+    scale_bytes: list[bytes],
+    runtime_registry,
+    return_scale_obj: bool = False,
+):
+    if scale_bytes == b"":
+        return None
+    if type_string == "scale_info::0":  # Is an AccountId
+        # Decode AccountId bytes to SS58 address
+        return ss58_encode(scale_bytes, SS58_FORMAT)
+    else:
+        obj = decode_list(type_string, runtime_registry, scale_bytes)
+    if return_scale_obj:
+        return [ScaleObj(x) for x in obj]
+    else:
+        return obj
+
+
+def decode_query_map(
+    result_group_changes,
+    prefix,
+    runtime: "Runtime",
+    param_types,
+    params,
+    value_type,
+    key_hashers,
+    ignore_decoding_errors,
+):
+    def concat_hash_len(key_hasher: str) -> int:
+        """
+        Helper function to avoid if statements
+        """
+        if key_hasher == "Blake2_128Concat":
+            return 16
+        elif key_hasher == "Twox64Concat":
+            return 8
+        elif key_hasher == "Identity":
+            return 0
+        else:
+            raise ValueError("Unsupported hash type")
+
+    hex_to_bytes_ = hex_to_bytes
+
+    result = []
+    # Determine type string
+    key_type_string_ = []
+    for n in range(len(params), len(param_types)):
+        key_type_string_.append(f"[u8; {concat_hash_len(key_hashers[n])}]")
+        key_type_string_.append(param_types[n])
+    key_type_string = f"({', '.join(key_type_string_)})"
+
+    pre_decoded_keys = []
+    pre_decoded_key_types = [key_type_string] * len(result_group_changes)
+    pre_decoded_values = []
+    pre_decoded_value_types = [value_type] * len(result_group_changes)
+
+    for item in result_group_changes:
+        pre_decoded_keys.append(bytes.fromhex(item[0][len(prefix) :]))
+        pre_decoded_values.append(hex_to_bytes_(item[1]))
+    all_decoded = _decode_scale_list_with_runtime(
+        pre_decoded_key_types + pre_decoded_value_types,
+        pre_decoded_keys + pre_decoded_values,
+        runtime.registry,
+    )
+    middl_index = len(all_decoded) // 2
+    decoded_keys = all_decoded[:middl_index]
+    decoded_values = [ScaleObj(x) for x in all_decoded[middl_index:]]
+    for dk, dv in zip(decoded_keys, decoded_values):
+        try:
+            # strip key_hashers to use as item key
+            if len(param_types) - len(params) == 1:
+                item_key = dk[1]
+            else:
+                item_key = tuple(
+                    dk[key + 1] for key in range(len(params), len(param_types) + 1, 2)
+                )
+
+        except Exception as _:
+            if not ignore_decoding_errors:
+                raise
+            item_key = None
+
+        item_value = dv
+        result.append([item_key, item_value])
+    return result

async_substrate_interface/utils/decoding_attempt.py

@@ -10,7 +10,7 @@ dependencies = [
   "wheel",
   "asyncstdlib~=3.13.0",
   "bittensor-wallet>=2.1.3",
-  "bt-decode==v0.5.0",
+  "bt-decode==v0.6.0",
   "scalecodec~=1.2.11",
   "websockets>=14.1",
   "xxhash"