Refactor generate_unique_id

async_substrate_interface/async_substrate.py

@@ -48,7 +48,7 @@
     SubstrateMixin,
     Preprocessed,
 )
-from async_substrate_interface.utils import hex_to_bytes, json, generate_unique_id
+from async_substrate_interface.utils import hex_to_bytes, json, get_next_id
 from async_substrate_interface.utils.decoding import (
     _determine_if_old_runtime_call,
     _bt_decode_to_dict_or_list,
@@ -620,7 +620,7 @@ async def send(self, payload: dict) -> int:
             id: the internal ID of the request (incremented int)
         """
         # async with self._lock:
-        original_id = generate_unique_id(json.dumps(payload))
+        original_id = get_next_id()
         # self._open_subscriptions += 1
         try:
             await self.ws.send(json.dumps({**payload, **{"id": original_id}}))

async_substrate_interface/sync_substrate.py

@@ -30,7 +30,7 @@
     Preprocessed,
     ScaleObj,
 )
-from async_substrate_interface.utils import hex_to_bytes, json, generate_unique_id
+from async_substrate_interface.utils import hex_to_bytes, json, get_next_id
 from async_substrate_interface.utils.decoding import (
     _determine_if_old_runtime_call,
     _bt_decode_to_dict_or_list,
@@ -1688,8 +1688,7 @@ def _make_rpc_request(
 
         ws = self.connect(init=False if attempt == 1 else True)
         for payload in payloads:
-            payload_str = json.dumps(payload["payload"])
-            item_id = generate_unique_id(payload_str)
+            item_id = get_next_id()
             ws.send(json.dumps({**payload["payload"], **{"id": item_id}}))
             request_manager.add_request(item_id, payload["id"])
 

async_substrate_interface/utils/__init__.py

@@ -1,10 +1,18 @@
 import importlib
-import hashlib
+from itertools import cycle
+import random
+import string
 
+id_cycle = cycle(range(1, 999))
 
-def generate_unique_id(item: str, length=10):
-    hashed_value = hashlib.sha256(item.encode()).hexdigest()
-    return hashed_value[:length]
+
+def get_next_id() -> str:
+    """
+    Generates a pseudo-random ID by returning the next int of a range from 1-998 prepended with
+    two random ascii characters.
+    """
+    random_letters = "".join(random.choices(string.ascii_letters, k=2))
+    return f"{random_letters}{next(id_cycle)}"
 
 
 def hex_to_bytes(hex_str: str) -> bytes:

tests/unit_tests/test_utils.py

@@ -0,0 +1,10 @@
+from async_substrate_interface.utils import get_next_id
+from string import ascii_letters
+
+
+def test_get_next_id():
+    next_id = get_next_id()
+    assert next_id[0] in ascii_letters
+    assert next_id[1] in ascii_letters
+    assert 0 < int(next_id[2:]) < 999
+    assert 3 <= len(next_id) <= 5