Metadata V15 support

docs/usage/call-runtime-apis.md

@@ -17,7 +17,7 @@ result = substrate.runtime_call("AccountNonceApi", "account_nonce", ["5GrwvaEF5z
 ```python
 runtime_calls = substrate.get_metadata_runtime_call_functions()
 #[
-#    <RuntimeCallDefinition(value={'description': 'The API to query account nonce (aka transaction index)', 'params': [{'name': 'account_id', 'type': 'AccountId'}], 'type': 'Index', 'api': 'AccountNonceApi', 'method': 'account_nonce'})>
+#   <RuntimeApiMetadataV14(value={'name': 'AccountNonceApi', 'methods': [{'name': 'account_nonce', 'inputs': [{'name': 'account_id', 'type': 'AccountId'}], 'output': 'Index', 'docs': ['The API to query account nonce (aka transaction index)']}], 'docs': []})>
 #    ...
 #]
 ```
@@ -28,5 +28,5 @@ A helper function to compose the parameters for this runtime API call
 ```python
 runtime_call = substrate.get_metadata_runtime_call_function("ContractsApi", "call")
 param_info = runtime_call.get_param_info()
-# ['AccountId', 'AccountId', 'u128', 'u64', (None, 'u128'), 'Bytes']
+# ['AccountId', 'AccountId', 'u128', (None, {'proof_size': 'u64', 'ref_time': 'u64'}), (None, 'u128'), 'Bytes']
 ```

requirements.txt

@@ -11,7 +11,7 @@ eth_utils>=1.3.0,<3
 pycryptodome>=3.11.0,<4
 PyNaCl>=1.0.1,<2
 
-scalecodec>=1.2.6,<1.3
+scalecodec==1.3.0a5
 py-sr25519-bindings>=0.2.0,<1
 py-ed25519-zebra-bindings>=1.0,<2
 py-bip39-bindings>=0.1.9,<1

setup.py

@@ -188,7 +188,7 @@
         'eth_utils>=1.3.0,<3',
         'pycryptodome>=3.11.0,<4',
         'PyNaCl>=1.0.1,<2',
-        'scalecodec>=1.2.6,<1.3',
+        'scalecodec==1.3.0a5',
         'py-sr25519-bindings>=0.2.0,<1',
         'py-ed25519-zebra-bindings>=1.0,<2',
         'py-bip39-bindings>=0.1.9,<1'

substrateinterface/base.py

@@ -29,7 +29,7 @@
 from websocket import create_connection, WebSocketConnectionClosedException
 
 from scalecodec.base import ScaleBytes, RuntimeConfigurationObject, ScaleType
-from scalecodec.types import GenericCall, GenericExtrinsic, Extrinsic, MultiAccountId, GenericRuntimeCallDefinition
+from scalecodec.types import GenericCall, GenericExtrinsic, Extrinsic, MultiAccountId, GenericRuntimeApiMethodMetadata
 from scalecodec.type_registry import load_type_registry_preset
 from scalecodec.updater import update_type_registries
 from .extensions import Extension
@@ -1248,24 +1248,25 @@ def runtime_call(self, api: str, method: str, params: Union[list, dict] = None,
         if params is None:
             params = {}
 
-        try:
-            runtime_call_def = self.runtime_config.type_registry["runtime_api"][api]['methods'][method]
-            runtime_api_types = self.runtime_config.type_registry["runtime_api"][api].get("types", {})
-        except KeyError:
-            raise ValueError(f"Runtime API Call '{api}.{method}' not found in registry")
+        runtime_api = self.metadata.get_runtime_api(api)
+
+        if not runtime_api:
+            raise ValueError(f"Runtime API '{api}' not found")
+
+        runtime_api_method = runtime_api.get_method(method)
 
-        if type(params) is list and len(params) != len(runtime_call_def['params']):
+        if not runtime_api_method:
+            raise ValueError(f"Runtime API method '{api}.{method}' not found")
+
+        if type(params) is list and len(params) != len(runtime_api_method.get_params()):
             raise ValueError(
                 f"Number of parameter provided ({len(params)}) does not "
-                f"match definition {len(runtime_call_def['params'])}"
+                f"match definition {len(runtime_api_method.get_params())}"
             )
 
-        # Add runtime API types to registry
-        self.runtime_config.update_type_registry_types(runtime_api_types)
-
         # Encode params
         param_data = ScaleBytes(bytes())
-        for idx, param in enumerate(runtime_call_def['params']):
+        for idx, param in enumerate(runtime_api_method.get_params()):
             scale_obj = self.runtime_config.create_scale_object(param['type'])
             if type(params) is list:
                 param_data += scale_obj.encode(params[idx])
@@ -1279,7 +1280,7 @@ def runtime_call(self, api: str, method: str, params: Union[list, dict] = None,
         result_data = self.rpc_request("state_call", [f'{api}_{method}', str(param_data), block_hash])
 
         # Decode result
-        result_obj = self.runtime_config.create_scale_object(runtime_call_def['type'])
+        result_obj = self.runtime_config.create_scale_object(runtime_api_method.get_return_type_string())
         result_obj.decode(ScaleBytes(result_data['result']), check_remaining=self.config.get('strict_scale_decode'))
 
         return result_obj
@@ -2250,53 +2251,51 @@ def get_metadata_error(self, module_name, error_name, block_hash=None):
                     if error_name == error.name:
                         return error
 
-    def get_metadata_runtime_call_functions(self) -> list:
+    def get_metadata_runtime_call_functions(self, block_hash: str = None) -> list:
         """
         Get a list of available runtime API calls
 
+        Parameters
+        ----------
+        block_hash: Optional block hash, when omitted the chain tip will be used
+
         Returns
         -------
         list
         """
-        self.init_runtime()
-        call_functions = []
-
-        for api, methods in self.runtime_config.type_registry["runtime_api"].items():
-            for method in methods["methods"].keys():
-                call_functions.append(self.get_metadata_runtime_call_function(api, method))
+        self.init_runtime(block_hash=block_hash)
 
-        return call_functions
+        return self.metadata.get_runtime_apis()
 
-    def get_metadata_runtime_call_function(self, api: str, method: str) -> GenericRuntimeCallDefinition:
+    def get_metadata_runtime_call_function(
+            self, api: str, method: str, block_hash: str = None
+    ) -> GenericRuntimeApiMethodMetadata:
         """
         Get details of a runtime API call
 
         Parameters
         ----------
         api: Name of the runtime API e.g. 'TransactionPaymentApi'
         method: Name of the method e.g. 'query_fee_details'
+        block_hash: Optional block hash, when omitted the chain tip will be used
 
         Returns
         -------
-        GenericRuntimeCallDefinition
+        GenericRuntimeApiMethodMetadata
         """
-        self.init_runtime()
+        self.init_runtime(block_hash=block_hash)
 
-        try:
-            runtime_call_def = self.runtime_config.type_registry["runtime_api"][api]['methods'][method]
-            runtime_call_def['api'] = api
-            runtime_call_def['method'] = method
-            runtime_api_types = self.runtime_config.type_registry["runtime_api"][api].get("types", {})
-        except KeyError:
-            raise ValueError(f"Runtime API Call '{api}.{method}' not found in registry")
+        runtime_api = self.metadata.get_runtime_api(api)
+
+        if not runtime_api:
+            raise ValueError(f"Runtime API '{api}' not found")
 
-        # Add runtime API types to registry
-        self.runtime_config.update_type_registry_types(runtime_api_types)
+        runtime_api_method = runtime_api.get_method(method)
 
-        runtime_call_def_obj = self.create_scale_object("RuntimeCallDefinition")
-        runtime_call_def_obj.encode(runtime_call_def)
+        if not runtime_api_method:
+            raise ValueError(f"Runtime API method '{api}.{method}' not found")
 
-        return runtime_call_def_obj
+        return runtime_api_method
 
     def __get_block_handler(self, block_hash: str, ignore_decoding_errors: bool = False, include_author: bool = False,
                             header_only: bool = False, finalized_only: bool = False,

test/test_runtime_call.py

@@ -80,13 +80,14 @@ def test_metadata_call_info(self):
         self.assertEqual('h256', param_info[0]['parent_hash'])
 
     def test_check_all_runtime_call_types(self):
-        runtime_calls = self.substrate.get_metadata_runtime_call_functions()
-        for runtime_call in runtime_calls:
-            param_info = runtime_call.get_param_info()
-            self.assertEqual(type(param_info), list)
-            result_obj = self.substrate.create_scale_object(runtime_call.value['type'])
-            info = result_obj.generate_type_decomposition()
-            self.assertIsNotNone(info)
+        runtime_apis = self.substrate.get_metadata_runtime_call_functions()
+        for api in runtime_apis:
+            for runtime_call in api.get_methods():
+                param_info = runtime_call.get_param_info()
+                self.assertEqual(type(param_info), list)
+                result_obj = self.substrate.create_scale_object(runtime_call.get_return_type_string())
+                info = result_obj.generate_type_decomposition()
+                self.assertIsNotNone(info)
 
     def test_unknown_runtime_call(self):
         with self.assertRaises(ValueError):