Skip to content

move metadata methods to SubstrateMixin #224

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 2 commits into from
Oct 24, 2025
Merged

move metadata methods to SubstrateMixin #224

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
98fe876
Moved all the metadata retrievals to shared functionality in the supe…
thewhaleking Oct 23, 2025
e1e37b3
Types!
thewhaleking Oct 23, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
150 changes: 22 additions & 128 deletions async_substrate_interface/async_substrate.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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)
Expand All @@ -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,
Expand Down Expand Up @@ -1662,28 +1649,15 @@ 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,
module_name: str,
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

Expand All @@ -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

Expand All @@ -1717,25 +1688,15 @@ 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,
api: str,
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.
Expand All @@ -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,
Expand Down Expand Up @@ -3422,24 +3362,15 @@ 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,
module_name: str,
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)
Expand All @@ -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,
Expand Down Expand Up @@ -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:
"""
Expand Down Expand Up @@ -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.
Expand Down Expand Up @@ -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]:
"""
Expand All @@ -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
Expand All @@ -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."""
Expand Down
Loading
Loading