diff --git a/async_substrate_interface/async_substrate.py b/async_substrate_interface/async_substrate.py index c2342e8..3fe188e 100644 --- a/async_substrate_interface/async_substrate.py +++ b/async_substrate_interface/async_substrate.py @@ -23,6 +23,7 @@ cast, ) +import scalecodec import websockets.exceptions from bt_decode import MetadataV15, PortableRegistry, decode as decode_by_type_string from scalecodec import GenericVariant @@ -1588,7 +1589,7 @@ async def retrieve_pending_extrinsics(self) -> list: async def get_metadata_storage_functions( self, block_hash: Optional[str] = None, runtime: Optional[Runtime] = None - ) -> list: + ) -> list[dict[str, Any]]: """ Retrieves a list of all storage functions in metadata active at given block_hash (or chaintip if block_hash and runtime are omitted) @@ -1603,21 +1604,7 @@ async def get_metadata_storage_functions( if runtime is None: runtime = await self.init_runtime(block_hash=block_hash) - storage_list = [] - - for module_idx, module in enumerate(runtime.metadata.pallets): - if module.storage: - for storage in module.storage: - storage_list.append( - self.serialize_storage_item( - storage_item=storage, - module=module, - spec_version_id=runtime.runtime_version, - runtime=runtime, - ) - ) - - return storage_list + return self._get_metadata_storage_functions(runtime=runtime) async def get_metadata_storage_function( self, @@ -1662,20 +1649,7 @@ async def get_metadata_errors( if runtime is None: runtime = await self.init_runtime(block_hash=block_hash) - error_list = [] - - for module_idx, module in enumerate(runtime.metadata.pallets): - if module.errors: - for error in module.errors: - error_list.append( - self.serialize_module_error( - module=module, - error=error, - spec_version=runtime.runtime_version, - ) - ) - - return error_list + return self._get_metadata_errors(runtime=runtime) async def get_metadata_error( self, @@ -1683,7 +1657,7 @@ async def get_metadata_error( error_name: str, block_hash: Optional[str] = None, runtime: Optional[Runtime] = None, - ): + ) -> Optional[scalecodec.GenericVariant]: """ Retrieves the details of an error for given module name, call function name and block_hash @@ -1699,16 +1673,13 @@ async def get_metadata_error( """ if runtime is None: runtime = await self.init_runtime(block_hash=block_hash) - - for module_idx, module in enumerate(runtime.metadata.pallets): - if module.name == module_name and module.errors: - for error in module.errors: - if error_name == error.name: - return error + return self._get_metadata_error( + module_name=module_name, error_name=error_name, runtime=runtime + ) async def get_metadata_runtime_call_functions( self, block_hash: Optional[str] = None, runtime: Optional[Runtime] = None - ) -> list[GenericRuntimeCallDefinition]: + ) -> list[scalecodec.GenericRuntimeCallDefinition]: """ Get a list of available runtime API calls @@ -1717,17 +1688,7 @@ async def get_metadata_runtime_call_functions( """ if runtime is None: runtime = await self.init_runtime(block_hash=block_hash) - call_functions = [] - - for api, methods in runtime.runtime_config.type_registry["runtime_api"].items(): - for method in methods["methods"].keys(): - call_functions.append( - await self.get_metadata_runtime_call_function( - api, method, runtime=runtime - ) - ) - - return call_functions + return self._get_metadata_runtime_call_functions(runtime=runtime) async def get_metadata_runtime_call_function( self, @@ -1735,7 +1696,7 @@ async def get_metadata_runtime_call_function( method: str, block_hash: Optional[str] = None, runtime: Optional[Runtime] = None, - ) -> GenericRuntimeCallDefinition: + ) -> scalecodec.GenericRuntimeCallDefinition: """ Get details of a runtime API call. If not supplying `block_hash` or `runtime`, the runtime of the current block will be used. @@ -1751,28 +1712,7 @@ async def get_metadata_runtime_call_function( """ if runtime is None: runtime = await self.init_runtime(block_hash=block_hash) - - try: - runtime_call_def = runtime.runtime_config.type_registry["runtime_api"][api][ - "methods" - ][method] - runtime_call_def["api"] = api - runtime_call_def["method"] = method - runtime_api_types = runtime.runtime_config.type_registry["runtime_api"][ - api - ].get("types", {}) - except KeyError: - raise ValueError(f"Runtime API Call '{api}.{method}' not found in registry") - - # Add runtime API types to registry - runtime.runtime_config.update_type_registry_types(runtime_api_types) - - runtime_call_def_obj = await self.create_scale_object( - "RuntimeCallDefinition", runtime=runtime - ) - runtime_call_def_obj.encode(runtime_call_def) - - return runtime_call_def_obj + return self._get_metadata_runtime_call_function(api, method, runtime) async def _get_block_handler( self, @@ -3422,16 +3362,7 @@ async def get_metadata_constants(self, block_hash=None) -> list[dict]: """ runtime = await self.init_runtime(block_hash=block_hash) - - constant_list = [] - - for module_idx, module in enumerate(runtime.metadata.pallets): - for constant in module.constants or []: - constant_list.append( - self.serialize_constant(constant, module, runtime.runtime_version) - ) - - return constant_list + return self._get_metadata_constants(runtime) async def get_metadata_constant( self, @@ -3439,7 +3370,7 @@ async def get_metadata_constant( constant_name: str, block_hash: Optional[str] = None, runtime: Optional[Runtime] = None, - ): + ) -> Optional[scalecodec.ScaleInfoModuleConstantMetadata]: """ Retrieves the details of a constant for given module name, call function name and block_hash (or chaintip if block_hash is omitted) @@ -3455,12 +3386,7 @@ async def get_metadata_constant( """ if runtime is None: runtime = await self.init_runtime(block_hash=block_hash) - - for module in runtime.metadata.pallets: - if module_name == module.name and module.constants: - for constant in module.constants: - if constant_name == constant.value["name"]: - return constant + return self._get_metadata_constant(module_name, constant_name, runtime) async def get_constant( self, @@ -3604,21 +3530,7 @@ async def get_metadata_modules(self, block_hash=None) -> list[dict[str, Any]]: List of metadata modules """ runtime = await self.init_runtime(block_hash=block_hash) - - return [ - { - "metadata_index": idx, - "module_id": module.get_identifier(), - "name": module.name, - "spec_version": runtime.runtime_version, - "count_call_functions": len(module.calls or []), - "count_storage_functions": len(module.storage or []), - "count_events": len(module.events or []), - "count_constants": len(module.constants or []), - "count_errors": len(module.errors or []), - } - for idx, module in enumerate(runtime.metadata.pallets) - ] + return self._get_metadata_modules(runtime) async def get_metadata_module(self, name, block_hash=None) -> ScaleType: """ @@ -4081,7 +3993,7 @@ async def result_handler(message: dict, subscription_id) -> tuple[dict, bool]: async def get_metadata_call_functions( self, block_hash: Optional[str] = None, runtime: Optional[Runtime] = None - ): + ) -> dict[str, dict[str, dict[str, dict[str, Union[str, int, list]]]]]: """ Retrieves calls functions for the metadata at the specified block_hash or runtime. If neither are specified, the metadata at chaintip is used. @@ -4133,12 +4045,9 @@ async def get_metadata_call_function( """ runtime = await self.init_runtime(block_hash=block_hash) - for pallet in runtime.metadata.pallets: - if pallet.name == module_name and pallet.calls: - for call in pallet.calls: - if call.name == call_function_name: - return call - return None + return self._get_metadata_call_function( + module_name, call_function_name, runtime + ) async def get_metadata_events(self, block_hash=None) -> list[dict]: """ @@ -4152,17 +4061,7 @@ async def get_metadata_events(self, block_hash=None) -> list[dict]: """ runtime = await self.init_runtime(block_hash=block_hash) - - event_list = [] - - for event_index, (module, event) in runtime.metadata.event_index.items(): - event_list.append( - self.serialize_module_event( - module, event, runtime.runtime_version, event_index - ) - ) - - return event_list + return self._get_metadata_events(runtime) async def get_metadata_event( self, module_name, event_name, block_hash=None @@ -4182,12 +4081,7 @@ async def get_metadata_event( """ runtime = await self.init_runtime(block_hash=block_hash) - - for pallet in runtime.metadata.pallets: - if pallet.name == module_name and pallet.events: - for event in pallet.events: - if event.name == event_name: - return event + return self._get_metadata_event(module_name, event_name, runtime) async def get_block_number(self, block_hash: Optional[str] = None) -> int: """Async version of `substrateinterface.base.get_block_number` method.""" diff --git a/async_substrate_interface/sync_substrate.py b/async_substrate_interface/sync_substrate.py index 373e8cd..fb9fad9 100644 --- a/async_substrate_interface/sync_substrate.py +++ b/async_substrate_interface/sync_substrate.py @@ -6,6 +6,7 @@ from typing import Optional, Union, Callable, Any from unittest.mock import MagicMock +import scalecodec from bt_decode import MetadataV15, PortableRegistry, decode as decode_by_type_string from scalecodec import ( GenericCall, @@ -1031,7 +1032,7 @@ def retrieve_pending_extrinsics(self) -> list: return extrinsics - def get_metadata_storage_functions(self, block_hash=None) -> list: + def get_metadata_storage_functions(self, block_hash=None) -> list[dict[str, Any]]: """ Retrieves a list of all storage functions in metadata active at given block_hash (or chaintip if block_hash is omitted) @@ -1042,22 +1043,8 @@ def get_metadata_storage_functions(self, block_hash=None) -> list: Returns: list of storage functions """ - self.init_runtime(block_hash=block_hash) - - storage_list = [] - - for module_idx, module in enumerate(self.metadata.pallets): - if module.storage: - for storage in module.storage: - storage_list.append( - self.serialize_storage_item( - storage_item=storage, - module=module, - spec_version_id=self.runtime.runtime_version, - ) - ) - - return storage_list + runtime = self.init_runtime(block_hash=block_hash) + return self._get_metadata_storage_functions(runtime=runtime) def get_metadata_storage_function(self, module_name, storage_name, block_hash=None): """ @@ -1088,24 +1075,13 @@ def get_metadata_errors(self, block_hash=None) -> list[dict[str, Optional[str]]] Returns: list of errors in the metadata """ - self.init_runtime(block_hash=block_hash) - - error_list = [] - - for module_idx, module in enumerate(self.runtime.metadata.pallets): - if module.errors: - for error in module.errors: - error_list.append( - self.serialize_module_error( - module=module, - error=error, - spec_version=self.runtime.runtime_version, - ) - ) + runtime = self.init_runtime(block_hash=block_hash) - return error_list + return self._get_metadata_errors(runtime=runtime) - def get_metadata_error(self, module_name, error_name, block_hash=None): + def get_metadata_error( + self, module_name: str, error_name: str, block_hash=None + ) -> Optional[scalecodec.GenericVariant]: """ Retrieves the details of an error for given module name, call function name and block_hash @@ -1118,37 +1094,26 @@ def get_metadata_error(self, module_name, error_name, block_hash=None): error """ - self.init_runtime(block_hash=block_hash) - - for module_idx, module in enumerate(self.runtime.metadata.pallets): - if module.name == module_name and module.errors: - for error in module.errors: - if error_name == error.name: - return error + runtime = self.init_runtime(block_hash=block_hash) + return self._get_metadata_error( + module_name=module_name, error_name=error_name, runtime=runtime + ) def get_metadata_runtime_call_functions( self, block_hash: Optional[str] = None - ) -> list[GenericRuntimeCallDefinition]: + ) -> list[scalecodec.GenericRuntimeCallDefinition]: """ Get a list of available runtime API calls Returns: list of runtime call functions """ - self.init_runtime(block_hash=block_hash) - 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) - ) - - return call_functions + runtime = self.init_runtime(block_hash=block_hash) + return self._get_metadata_runtime_call_functions(runtime=runtime) def get_metadata_runtime_call_function( self, api: str, method: str, block_hash: Optional[str] = None - ) -> GenericRuntimeCallDefinition: + ) -> scalecodec.GenericRuntimeCallDefinition: """ Get details of a runtime API call @@ -1160,27 +1125,9 @@ def get_metadata_runtime_call_function( Returns: runtime call function """ - 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") - - # Add runtime API types to registry - self.runtime_config.update_type_registry_types(runtime_api_types) - - runtime_call_def_obj = self.create_scale_object("RuntimeCallDefinition") - runtime_call_def_obj.encode(runtime_call_def) + runtime = self.init_runtime(block_hash=block_hash) - return runtime_call_def_obj + return self._get_metadata_runtime_call_function(api, method, runtime) def _get_block_handler( self, @@ -2716,17 +2663,11 @@ def get_metadata_constants(self, block_hash=None) -> list[dict]: runtime = self.init_runtime(block_hash=block_hash) - constant_list = [] - - for module_idx, module in enumerate(self.metadata.pallets): - for constant in module.constants or []: - constant_list.append( - self.serialize_constant(constant, module, runtime.runtime_version) - ) - - return constant_list + return self._get_metadata_constants(runtime) - def get_metadata_constant(self, module_name, constant_name, block_hash=None): + def get_metadata_constant( + self, module_name, constant_name, block_hash=None + ) -> Optional[scalecodec.ScaleInfoModuleConstantMetadata]: """ Retrieves the details of a constant for given module name, call function name and block_hash (or chaintip if block_hash is omitted) @@ -2739,13 +2680,8 @@ def get_metadata_constant(self, module_name, constant_name, block_hash=None): Returns: MetadataModuleConstants """ - self.init_runtime(block_hash=block_hash) - - for module in self.runtime.metadata.pallets: - if module_name == module.name and module.constants: - for constant in module.constants: - if constant_name == constant.value["name"]: - return constant + runtime = self.init_runtime(block_hash=block_hash) + return self._get_metadata_constant(module_name, constant_name, runtime) def get_constant( self, @@ -2881,22 +2817,8 @@ def get_metadata_modules(self, block_hash=None) -> list[dict[str, Any]]: Returns: List of metadata modules """ - self.init_runtime(block_hash=block_hash) - - return [ - { - "metadata_index": idx, - "module_id": module.get_identifier(), - "name": module.name, - "spec_version": self.runtime.runtime_version, - "count_call_functions": len(module.calls or []), - "count_storage_functions": len(module.storage or []), - "count_events": len(module.events or []), - "count_constants": len(module.constants or []), - "count_errors": len(module.errors or []), - } - for idx, module in enumerate(self.metadata.pallets) - ] + runtime = self.init_runtime(block_hash=block_hash) + return self._get_metadata_modules(runtime) def get_metadata_module(self, name, block_hash=None) -> ScaleType: """ @@ -3291,7 +3213,9 @@ def result_handler(message: dict, subscription_id) -> tuple[dict, bool]: return result - def get_metadata_call_functions(self, block_hash: Optional[str] = None): + def get_metadata_call_functions( + self, block_hash: Optional[str] = None + ) -> dict[str, dict[str, dict[str, dict[str, Union[str, int, list]]]]]: """ Retrieves calls functions for the metadata at the specified block_hash. If not specified, the metadata at chaintip is used. @@ -3339,14 +3263,11 @@ def get_metadata_call_function( Returns: The dict-like call definition, if found. None otherwise. """ - self.init_runtime(block_hash=block_hash) + runtime = self.init_runtime(block_hash=block_hash) - for pallet in self.runtime.metadata.pallets: - if pallet.name == module_name and pallet.calls: - for call in pallet.calls: - if call.name == call_function_name: - return call - return None + return self._get_metadata_call_function( + module_name, call_function_name, runtime + ) def get_metadata_events(self, block_hash=None) -> list[dict]: """ @@ -3360,20 +3281,10 @@ def get_metadata_events(self, block_hash=None) -> list[dict]: """ runtime = self.init_runtime(block_hash=block_hash) - - event_list = [] - - for event_index, (module, event) in self.metadata.event_index.items(): - event_list.append( - self.serialize_module_event( - module, event, runtime.runtime_version, event_index - ) - ) - - return event_list + return self._get_metadata_events(runtime) def get_metadata_event( - self, module_name, event_name, block_hash=None + self, module_name: str, event_name: str, block_hash=None ) -> Optional[Any]: """ Retrieves the details of an event for given module name, call function name and block_hash @@ -3390,12 +3301,7 @@ def get_metadata_event( """ runtime = self.init_runtime(block_hash=block_hash) - - for pallet in runtime.metadata.pallets: - if pallet.name == module_name and pallet.events: - for event in pallet.events: - if event.name == event_name: - return 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.""" diff --git a/async_substrate_interface/types.py b/async_substrate_interface/types.py index 008dd48..a627132 100644 --- a/async_substrate_interface/types.py +++ b/async_substrate_interface/types.py @@ -676,8 +676,8 @@ def is_valid_ss58_address(self, value: str) -> bool: def serialize_storage_item( self, - storage_item: ScaleType, - module: str, + storage_item: scalecodec.ScaleInfoStorageEntryMetadata, + module: scalecodec.ScaleInfoPalletMetadata, spec_version_id: int, runtime: Optional[Runtime] = None, ) -> dict: @@ -1004,7 +1004,9 @@ def generate_multisig_account( return multi_sig_account @staticmethod - def _get_metadata_call_functions(runtime: Runtime): + def _get_metadata_call_functions( + runtime: Runtime, + ) -> dict[str, dict[str, dict[str, dict[str, Union[str, int, list]]]]]: """ See subclass `get_metadata_call_functions` for documentation. """ @@ -1020,3 +1022,193 @@ def _get_metadata_call_functions(runtime: Runtime): field["_docs"] = " ".join(field_docs) data[pallet.name][call.name][field["name"]] = field return data + + @staticmethod + def _get_metadata_call_function( + module_name: str, call_function_name: str, runtime: Runtime + ) -> Optional[scalecodec.GenericVariant]: + """ + See subclass `get_metadata_call_function` for documentation. + """ + for pallet in runtime.metadata.pallets: + if pallet.name == module_name and pallet.calls: + for call in pallet.calls: + if call.name == call_function_name: + return call + return None + + def _get_metadata_events(self, runtime: Runtime) -> list[dict]: + """ + See subclass `get_metadata_events` for documentation. + """ + event_list = [] + + for event_index, (module, event) in runtime.metadata.event_index.items(): + event_list.append( + self.serialize_module_event( + module, event, runtime.runtime_version, event_index + ) + ) + + return event_list + + @staticmethod + def _get_metadata_event( + module_name: str, event_name: str, runtime: Runtime + ) -> Optional[scalecodec.GenericScaleInfoEvent]: + """ + See subclass `get_metadata_event` for documentation. + """ + for pallet in runtime.metadata.pallets: + if pallet.name == module_name and pallet.events: + for event in pallet.events: + if event.name == event_name: + return event + return None + + def _get_metadata_constants(self, runtime: Runtime) -> list[dict]: + """ + See subclass `get_metadata_constants` for documentation. + """ + constant_list = [] + + for module_idx, module in enumerate(runtime.metadata.pallets): + for constant in module.constants or []: + constant_list.append( + self.serialize_constant(constant, module, runtime.runtime_version) + ) + + return constant_list + + @staticmethod + def _get_metadata_constant( + module_name: str, constant_name: str, runtime: Runtime + ) -> Optional[scalecodec.ScaleInfoModuleConstantMetadata]: + """ + See subclass `get_metadata_constant` for documentation. + """ + for module in runtime.metadata.pallets: + if module_name == module.name and module.constants: + for constant in module.constants: + if constant_name == constant.value["name"]: + return constant + return None + + @staticmethod + def _get_metadata_modules(runtime: Runtime) -> list[dict[str, Any]]: + """ + See subclass `get_metadata_modules` for documentation. + """ + return [ + { + "metadata_index": idx, + "module_id": module.get_identifier(), + "name": module.name, + "spec_version": runtime.runtime_version, + "count_call_functions": len(module.calls or []), + "count_storage_functions": len(module.storage or []), + "count_events": len(module.events or []), + "count_constants": len(module.constants or []), + "count_errors": len(module.errors or []), + } + for idx, module in enumerate(runtime.metadata.pallets) + ] + + def _get_metadata_storage_functions(self, runtime: Runtime) -> list[dict[str, Any]]: + """ + See subclass `get_metadata_storage_functions` for documentation. + """ + storage_list = [] + + for module_idx, module in enumerate(runtime.metadata.pallets): + if module.storage: + for storage in module.storage: + storage_list.append( + self.serialize_storage_item( + storage_item=storage, + module=module, + spec_version_id=self.runtime.runtime_version, + runtime=runtime, + ) + ) + + return storage_list + + def _get_metadata_errors(self, runtime: Runtime) -> list[dict[str, Optional[str]]]: + """ + See subclass `get_metadata_errors` for documentation. + """ + error_list = [] + + for module_idx, module in enumerate(runtime.metadata.pallets): + if module.errors: + for error in module.errors: + error_list.append( + self.serialize_module_error( + module=module, + error=error, + spec_version=runtime.runtime_version, + ) + ) + + return error_list + + @staticmethod + def _get_metadata_error( + module_name: str, error_name: str, runtime: Runtime + ) -> Optional[scalecodec.GenericVariant]: + """ + See subclass `get_metadata_error` for documentation. + """ + for module_idx, module in enumerate(runtime.metadata.pallets): + if module.name == module_name and module.errors: + for error in module.errors: + if error_name == error.name: + return error + return None + + @staticmethod + def _get_metadata_runtime_call_function( + api: str, method: str, runtime: Runtime + ) -> scalecodec.GenericRuntimeCallDefinition: + """ + See subclass `get_metadata_runtime_call_function` for documentation. + """ + try: + runtime_call_def = runtime.runtime_config.type_registry["runtime_api"][api][ + "methods" + ][method] + runtime_call_def["api"] = api + runtime_call_def["method"] = method + runtime_api_types = runtime.runtime_config.type_registry["runtime_api"][ + api + ].get("types", {}) + except KeyError: + raise ValueError(f"Runtime API Call '{api}.{method}' not found in registry") + + # Add runtime API types to registry + runtime.runtime_config.update_type_registry_types(runtime_api_types) + runtime_call_def_obj = runtime.runtime_config.create_scale_object( + "RuntimeCallDefinition" + ) + runtime_call_def_obj.encode(runtime_call_def) + + return runtime_call_def_obj + + def _get_metadata_runtime_call_functions( + self, runtime: Runtime + ) -> list[scalecodec.GenericRuntimeCallDefinition]: + """ + See subclass `get_metadata_runtime_call_functions` for documentation. + """ + call_functions = [] + + for api, methods in runtime.runtime_config.type_registry["runtime_api"].items(): + for method in methods["methods"].keys(): + call_functions.append( + self._get_metadata_runtime_call_function( + api=api, method=method, runtime=runtime + ) + ) + + return call_functions