Consolidate code.

Author: thewhaleking

async_substrate_interface/async_substrate.py

@@ -807,15 +807,7 @@ async def load_registry(self):
             metadata_option_bytes
         )
         self.registry = PortableRegistry.from_metadata_v15(self.metadata_v15)
-        registry_type_map = {}
-        for i in json.loads(self.registry.registry)["types"]:
-            for variants in (
-                i.get("type").get("def", {}).get("variant", {}).get("variants", [{}])
-            ):
-                for field in variants.get("fields", [{}]):
-                    if field.get("type") and field.get("typeName"):
-                        registry_type_map[field["typeName"]] = field["type"]
-        self.registry_type_map = registry_type_map
+        self._load_registry_type_map()
 
     async def _load_registry_at_block(self, block_hash: str) -> MetadataV15:
         # Should be called for any block that fails decoding.

async_substrate_interface/sync_substrate.py

@@ -611,15 +611,7 @@ def load_registry(self):
             metadata_option_bytes
         )
         self.registry = PortableRegistry.from_metadata_v15(self.metadata_v15)
-        registry_type_map = {}
-        for i in json.loads(self.registry.registry)["types"]:
-            for variants in (
-                i.get("type").get("def", {}).get("variant", {}).get("variants", [{}])
-            ):
-                for field in variants.get("fields", [{}]):
-                    if field.get("type") and field.get("typeName"):
-                        registry_type_map[field["typeName"]] = field["type"]
-        self.registry_type_map = registry_type_map
+        self._load_registry_type_map()
 
     def _load_registry_at_block(self, block_hash: str) -> MetadataV15:
         # Should be called for any block that fails decoding.

async_substrate_interface/types.py

@@ -13,6 +13,8 @@
 from scalecodec.type_registry import load_type_registry_preset
 from scalecodec.types import GenericCall, ScaleType
 
+from .utils import json
+
 
 class RuntimeCache:
     blocks: dict[int, "Runtime"]
@@ -346,6 +348,7 @@ class SubstrateMixin(ABC):
     ss58_format: Optional[int]
     ws_max_size = 2**32
     registry_type_map: dict[str, int]
+    metadata_v15 = None
 
     @property
     def chain(self):
@@ -601,6 +604,17 @@ def serialize_module_error(module, error, spec_version) -> dict:
             "spec_version": spec_version,
         }
 
+    def _load_registry_type_map(self):
+        registry_type_map = {}
+        for i in json.loads(self.registry.registry)["types"]:
+            for variants in (
+                i.get("type").get("def", {}).get("variant", {}).get("variants", [{}])
+            ):
+                for field in variants.get("fields", [{}]):
+                    if field.get("type") and field.get("typeName"):
+                        registry_type_map[field["typeName"]] = field["type"]
+        self.registry_type_map = registry_type_map
+
     def reload_type_registry(
         self, use_remote_preset: bool = True, auto_discover: bool = True
     ):