Skip to content

Base

SubstrateInterface

__init__(url = None, websocket = None, ss58_format = None, type_registry = None, type_registry_preset = None, cache_region = None, runtime_config = None, use_remote_preset = False, ws_options = None, auto_discover = True, auto_reconnect = True)

A specialized class in interfacing with a Substrate node.

Parameters:

Name Type Description Default
url None
ss58_format None
type_registry None
type_registry_preset None
cache_region None
use_remote_preset False
ws_options None

connect_websocket()

(Re)creates the websocket connection, if the URL contains a 'ws' or 'wss' scheme

close()

Cleans up resources for this instance like active websocket connection and active extensions

debug_message(message: str) staticmethod

Submits a message to the debug logger

Parameters:

Name Type Description Default
message str required

supports_rpc_method(name: str) -> bool

Check if substrate RPC supports given method

Parameters:

Name Type Description Default
name str required

Returns:

Type Description
bool

rpc_request(method, params, result_handler = None)

Method that handles the actual RPC request to the Substrate node. The other implemented functions eventually use this method to perform the request.

Parameters:

Name Type Description Default
result_handler None
method required
params required

Returns:

Type Description
a dict with the parsed result of the request.

implements_scaleinfo() -> Optional[bool]

Returns True if current runtime implementation a PortableRegistry (MetadataV14 and higher)

Returns:

Type Description
bool

get_chain_head()

A pass-though to existing JSONRPC method chain_getHead

get_chain_finalised_head()

A pass-though to existing JSONRPC method chain_getFinalizedHead

get_block_hash(block_id: int = None) -> str

A pass-though to existing JSONRPC method chain_getBlockHash

Parameters:

Name Type Description Default
block_id int None

get_block_number(block_hash: str) -> int

A convenience method to get the block number for given block_hash

Parameters:

Name Type Description Default
block_hash str required

get_block_metadata(block_hash = None, decode = True)

A pass-though to existing JSONRPC method state_getMetadata.

Parameters:

Name Type Description Default
block_hash None
decode True

get_storage_by_key(block_hash: str, storage_key: str)

A pass-though to existing JSONRPC method state_getStorage

Parameters:

Name Type Description Default
block_hash str required
storage_key str required

get_block_runtime_version(block_hash)

Retrieve the runtime version id of given block_hash

Parameters:

Name Type Description Default
block_hash required

generate_storage_hash(storage_module: str, storage_function: str, params: list = None, hashers: list = None)

Generate a storage key for given module/function

Parameters:

Name Type Description Default
storage_module str required
storage_function str required
params list None
hashers list None

Returns:

Type Description
str Hexstring respresentation of the storage key

init_runtime(block_hash = None, block_id = None)

This method is used by all other methods that deals with metadata and types defined in the type registry. It optionally retrieves the block_hash when block_id is given and sets the applicable metadata for that block_hash. Also it applies all the versioned types at the time of the block_hash.

Because parsing of metadata and type registry is quite heavy, the result will be cached per runtime id. In the future there could be support for caching backends like Redis to make this cache more persistent.

Parameters:

Name Type Description Default
block_hash None
block_id None

query_map(module: str, storage_function: str, params: Optional[list] = None, block_hash: str = None, max_results: int = None, start_key: str = None, page_size: int = 100, ignore_decoding_errors: bool = True) -> QueryMapResult

Iterates over all key-pairs located at the given module and storage_function. The storage item must be a map.

Example:

1
2
3
4
result = substrate.query_map('System', 'Account', max_results=100)

for account, account_info in result:
    print(f"Free balance of account '{account.value}': {account_info.value['data']['free']}")

Parameters:

Name Type Description Default
module str required
storage_function str required
params Optional[list] None
block_hash str None
max_results int None
start_key str None
page_size int 100
ignore_decoding_errors bool True

Returns:

Type Description
QueryMapResult

query_multi(storage_keys: List[StorageKey], block_hash: Optional[str] = None) -> list

Query multiple storage keys in one request.

Example:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
storage_keys = [
    substrate.create_storage_key(
        "System", "Account", ["F4xQKRUagnSGjFqafyhajLs94e7Vvzvr8ebwYJceKpr8R7T"]
    ),
    substrate.create_storage_key(
        "System", "Account", ["GSEX8kR4Kz5UZGhvRUCJG93D5hhTAoVZ5tAe6Zne7V42DSi"]
    )
]

result = substrate.query_multi(storage_keys)

Parameters:

Name Type Description Default
storage_keys List[StorageKey] required
block_hash Optional[str] None

Returns:

Type Description
list of

query(module: str, storage_function: str, params: list = None, block_hash: str = None, subscription_handler: callable = None, raw_storage_key: bytes = None) -> ScaleType

Retrieves the storage entry for given module, function and optional parameters at given block hash.

When a subscription_handler callback function is passed, a subscription will be maintained as long as this handler doesn't return a value.

Example of subscription handler:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def subscription_handler(obj, update_nr, subscription_id):

    if update_nr == 0:
        print('Initial data:', obj.value)

    if update_nr > 0:
        # Do something with the update
        print('data changed:', obj.value)

    # The execution will block until an arbitrary value is returned, which will be the result of the `query`
    if update_nr > 1:
        return obj

Parameters:

Name Type Description Default
module str required
storage_function str required
params list None
block_hash str None
subscription_handler callable None
raw_storage_key bytes None

Returns:

Type Description
ScaleType

__query_well_known(name: str, block_hash: str) -> ScaleType

Query well-known storage keys as defined in Substrate

Parameters:

Name Type Description Default
name str required
block_hash str required

Returns:

Type Description
Optional[ScaleType]

create_storage_key(pallet: str, storage_function: str, params: Optional[list] = None) -> StorageKey

Create a StorageKey instance providing storage function details. See subscribe_storage().

Parameters:

Name Type Description Default
pallet str required
storage_function str required
params Optional[list] None

Returns:

Type Description
StorageKey

subscribe_storage(storage_keys: List[StorageKey], subscription_handler: callable)

Subscribe to provided storage_keys and keep tracking until subscription_handler returns a value

Example of a StorageKey:

1
2
3
StorageKey.create_from_storage_function(
    "System", "Account", ["5GrwvaEF5zXb26Fz9rcQpDWS57CtERHpNehXCPcNoHGKutQY"]
)

Example of a subscription handler:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
def subscription_handler(storage_key, obj, update_nr, subscription_id):

    if update_nr == 0:
        print('Initial data:', storage_key, obj.value)

    if update_nr > 0:
        # Do something with the update
        print('data changed:', storage_key, obj.value)

    # The execution will block until an arbitrary value is returned, which will be the result of the function
    if update_nr > 1:
        return obj

Parameters:

Name Type Description Default
storage_keys List[StorageKey] required
subscription_handler callable required

retrieve_pending_extrinsics() -> list

Retrieves and decodes pending extrinsics from the node's transaction pool

Returns:

Type Description
list of extrinsics

runtime_call(api: str, method: str, params: Union[list, dict] = None) -> ScaleType

Calls a runtime API method

Parameters:

Name Type Description Default
api str required
method str required
params Union[list, dict] None

Returns:

Type Description
ScaleType

get_events(block_hash: str = None) -> list

Convenience method to get events for a certain block (storage call for module 'System' and function 'Events')

Parameters:

Name Type Description Default
block_hash str None

Returns:

Type Description
list

get_metadata(block_hash = None)

Returns MetadataVersioned object for given block_hash or chaintip if block_hash is omitted

Parameters:

Name Type Description Default
block_hash None

Returns:

Type Description
MetadataVersioned

get_runtime_metadata(block_hash = None)

Retrieves and decodes the metadata for given block or chaintip if block_hash is omitted.

Parameters:

Name Type Description Default
block_hash None

create_scale_object(type_string: str, data: ScaleBytes = None, block_hash: str = None, **kwargs) -> ScaleType

Convenience method to create a SCALE object of type type_string, this will initialize the runtime automatically at moment of block_hash, or chain tip if omitted.

Parameters:

Name Type Description Default
type_string str required
data ScaleBytes None
block_hash str None
kwargs {}

Returns:

Type Description
ScaleType

compose_call(call_module: str, call_function: str, call_params: dict = None, block_hash: str = None)

Composes a call payload which can be used in an extrinsic.

Parameters:

Name Type Description Default
call_module str required
call_function str required
call_params dict None
block_hash str None

Returns:

Type Description
GenericCall

get_account_nonce(account_address) -> int

Returns current nonce for given account address

Parameters:

Name Type Description Default
account_address required

Returns:

Type Description
int

create_signed_extrinsic(call: GenericCall, keypair: Keypair, era: dict = None, nonce: int = None, tip: int = 0, tip_asset_id: int = None, signature: Union[bytes, str] = None) -> GenericExtrinsic

Creates a extrinsic signed by given account details

Parameters:

Name Type Description Default
call GenericCall required
keypair Keypair required
era dict None
nonce int None
tip int 0
tip_asset_id int None
signature Union[bytes, str] None

Returns:

Type Description
GenericExtrinsic The signed Extrinsic

create_unsigned_extrinsic(call: GenericCall) -> GenericExtrinsic

Create unsigned extrinsic for given Call

Parameters:

Name Type Description Default
call GenericCall required

Returns:

Type Description
GenericExtrinsic

generate_multisig_account(signatories: list, threshold: int) -> MultiAccountId

Generate deterministic Multisig account with supplied signatories and threshold

Parameters:

Name Type Description Default
signatories list required
threshold int required

Returns:

Type Description
MultiAccountId

create_multisig_extrinsic(call: GenericCall, keypair: Keypair, multisig_account: MultiAccountId, max_weight: Optional[Union[dict, int]] = None, era: dict = None, nonce: int = None, tip: int = 0, tip_asset_id: int = None, signature: Union[bytes, str] = None) -> GenericExtrinsic

Create a Multisig extrinsic that will be signed by one of the signatories. Checks on-chain if the threshold of the multisig account is reached and try to execute the call accordingly.

Parameters:

Name Type Description Default
call GenericCall required
keypair Keypair required
multisig_account MultiAccountId required
max_weight Optional[Union[dict, int]] None
era dict None
nonce int None
tip int 0
tip_asset_id int None
signature Union[bytes, str] None

Returns:

Type Description
GenericExtrinsic

submit_extrinsic(extrinsic: GenericExtrinsic, wait_for_inclusion: bool = False, wait_for_finalization: bool = False) -> ExtrinsicReceipt

Submit an extrinsic to the connected node, with the possibility to wait until the extrinsic is included in a block and/or the block is finalized. The receipt returned provided information about the block and triggered events

Parameters:

Name Type Description Default
extrinsic GenericExtrinsic required
wait_for_inclusion bool False
wait_for_finalization bool False

Returns:

Type Description
ExtrinsicReceipt

get_payment_info(call: GenericCall, keypair: Keypair)

Retrieves fee estimation via RPC for given extrinsic

Parameters:

Name Type Description Default
call GenericCall required
keypair Keypair required

Returns:

Type Description
Dict with payment info

E.g. {'class': 'normal', 'partialFee': 151000000, 'weight': {'ref_time': 143322000}}

get_type_registry(block_hash: str = None, max_recursion: int = 4) -> dict

Generates an exhaustive list of which RUST types exist in the runtime specified at given block_hash (or chaintip if block_hash is omitted)

MetadataV14 or higher is required.

Parameters:

Name Type Description Default
block_hash str None
max_recursion int 4

Returns:

Type Description
dict

get_type_definition(type_string: str, block_hash: str = None)

Retrieves SCALE encoding specifications of given type_string

Parameters:

Name Type Description Default
type_string str required
block_hash str None

get_metadata_modules(block_hash = None)

Retrieves a list of modules in metadata for given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
block_hash None

get_metadata_module(name, block_hash = None)

Retrieves modules in metadata by name for given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
name required
block_hash None

Returns:

Type Description
MetadataModule

get_metadata_call_functions(block_hash = None) -> list

Retrieves a list of all call functions in metadata active for given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
block_hash None

Returns:

Type Description
list

get_metadata_call_function(module_name: str, call_function_name: str, block_hash: str = None)

Retrieves the details of a call function given module name, call function name and block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
module_name str required
call_function_name str required
block_hash str None

get_metadata_events(block_hash = None) -> list

Retrieves a list of all events in metadata active for given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
block_hash None

Returns:

Type Description
list

get_metadata_event(module_name, event_name, block_hash = None)

Retrieves the details of an event for given module name, call function name and block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
module_name required
event_name required
block_hash None

get_metadata_constants(block_hash = None) -> list

Retrieves a list of all constants in metadata active at given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
block_hash None

Returns:

Type Description
list

get_metadata_constant(module_name, constant_name, block_hash = None)

Retrieves the details of a constant for given module name, call function name and block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
module_name required
constant_name required
block_hash None

Returns:

Type Description
MetadataModuleConstants

get_constant(module_name, constant_name, block_hash = None) -> Optional[ScaleType]

Returns the decoded ScaleType object of the constant for given module name, call function name and block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
module_name required
constant_name required
block_hash None

Returns:

Type Description
ScaleType

get_metadata_storage_functions(block_hash = None) -> list

Retrieves a list of all storage functions in metadata active at given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
block_hash None

Returns:

Type Description
list

get_metadata_storage_function(module_name, storage_name, block_hash = None)

Retrieves the details of a storage function for given module name, call function name and block_hash

Parameters:

Name Type Description Default
module_name required
storage_name required
block_hash None

get_metadata_errors(block_hash = None) -> list

Retrieves a list of all errors in metadata active at given block_hash (or chaintip if block_hash is omitted)

Parameters:

Name Type Description Default
block_hash None

Returns:

Type Description
list

get_metadata_error(module_name, error_name, block_hash = None)

Retrieves the details of an error for given module name, call function name and block_hash

Parameters:

Name Type Description Default
module_name required
error_name required
block_hash None

get_metadata_runtime_call_functions() -> list

Get a list of available runtime API calls

Returns:

Type Description
list

get_metadata_runtime_call_function(api: str, method: str) -> GenericRuntimeCallDefinition

Get details of a runtime API call

Parameters:

Name Type Description Default
api str required
method str required

Returns:

Type Description
GenericRuntimeCallDefinition

get_block(block_hash: str = None, block_number: int = None, ignore_decoding_errors: bool = False, include_author: bool = False, finalized_only: bool = False) -> Optional[dict]

Retrieves a block and decodes its containing extrinsics and log digest items. If block_hash and block_number is omited the chain tip will be retrieve, or the finalized head if finalized_only is set to true.

Either block_hash or block_number should be set, or both omitted.

Parameters:

Name Type Description Default
block_hash str None
block_number int None
ignore_decoding_errors bool False
include_author bool False
finalized_only bool False

Returns:

Type Description
A dict containing the extrinsic and digest logs data

get_block_header(block_hash: str = None, block_number: int = None, ignore_decoding_errors: bool = False, include_author: bool = False, finalized_only: bool = False)

Retrieves a block header and decodes its containing log digest items. If block_hash and block_number is omited the chain tip will be retrieve, or the finalized head if finalized_only is set to true.

Either block_hash or block_number should be set, or both omitted.

See get_block() to also include the extrinsics in the result

Parameters:

Name Type Description Default
block_hash str None
block_number int None
ignore_decoding_errors bool False
include_author bool False
finalized_only bool False

Returns:

Type Description
A dict containing the header and digest logs data

subscribe_block_headers(subscription_handler: callable, ignore_decoding_errors: bool = False, include_author: bool = False, finalized_only = False)

Subscribe to new block headers as soon as they are available. The callable subscription_handler will be executed when a new block is available and execution will block until subscription_handler will return a result other than None.

Example:

1
2
3
4
5
6
7
8
9
def subscription_handler(obj, update_nr, subscription_id):

    print(f"New block #{obj['header']['number']} produced by {obj['header']['author']}")

    if update_nr > 10
      return {'message': 'Subscription will cancel when a value is returned', 'updates_processed': update_nr}


result = substrate.subscribe_block_headers(subscription_handler, include_author=True)

Parameters:

Name Type Description Default
subscription_handler callable required
ignore_decoding_errors bool False
include_author bool False
finalized_only False

Returns:

Type Description
Value return by

retrieve_extrinsic_by_identifier(extrinsic_identifier: str) -> ExtrinsicReceipt

Retrieve an extrinsic by its identifier in format "[block_number]-[extrinsic_index]" e.g. 333456-4

Parameters:

Name Type Description Default
extrinsic_identifier str required

Returns:

Type Description
ExtrinsicReceipt

retrieve_extrinsic_by_hash(block_hash: str, extrinsic_hash: str) -> ExtrinsicReceipt

Retrieve an extrinsic by providing the block_hash and the extrinsic hash

Parameters:

Name Type Description Default
block_hash str required
extrinsic_hash str required

Returns:

Type Description
ExtrinsicReceipt

get_extrinsics(block_hash: str = None, block_number: int = None) -> list

Return extrinsics for given block_hash or block_number

Parameters:

Name Type Description Default
block_hash str None
block_number int None

decode_scale(type_string, scale_bytes, block_hash = None, return_scale_obj = False)

Helper function to decode arbitrary SCALE-bytes (e.g. 0x02000000) according to given RUST type_string (e.g. BlockNumber). The relevant versioning information of the type (if defined) will be applied if block_hash is set

Parameters:

Name Type Description Default
type_string required
scale_bytes required
block_hash None
return_scale_obj False

encode_scale(type_string, value, block_hash = None) -> ScaleBytes

Helper function to encode arbitrary data into SCALE-bytes for given RUST type_string

Parameters:

Name Type Description Default
type_string required
value required
block_hash None

Returns:

Type Description
ScaleBytes

ss58_encode(public_key: Union[str, bytes], ss58_format: int = None) -> str

Helper function to encode a public key to SS58 address.

If no target ss58_format is provided, it will default to the ss58 format of the network it's connected to.

Parameters:

Name Type Description Default
public_key Union[str, bytes] required
ss58_format int None

Returns:

Type Description
str containing the SS58 address

ss58_decode(ss58_address: str) -> str

Helper function to decode a SS58 address to a public key

Parameters:

Name Type Description Default
ss58_address str required

Returns:

Type Description
str containing the hex representation of the public key

is_valid_ss58_address(value: str) -> bool

Helper function to validate given value as ss58_address for current network/ss58_format

Parameters:

Name Type Description Default
value str required

Returns:

Type Description
bool

serialize_storage_item(storage_item, module, spec_version_id) -> dict

Helper function to serialize a storage item

Parameters:

Name Type Description Default
storage_item required
module required
spec_version_id required

Returns:

Type Description
dict

serialize_constant(constant, module, spec_version_id) -> dict

Helper function to serialize a constant

Parameters:

Name Type Description Default
constant required
module required
spec_version_id required

Returns:

Type Description
dict

serialize_module_call(module, call, spec_version, call_index = None) -> dict

Helper function to serialize a call function

Parameters:

Name Type Description Default
module required
call required
spec_version required
call_index None

Returns:

Type Description
dict

serialize_module_event(module, event, spec_version, event_index) -> dict

Helper function to serialize an event

Parameters:

Name Type Description Default
module required
event required
spec_version required
event_index required

Returns:

Type Description
dict

serialize_module_error(module, error, spec_version) -> dict

Helper function to serialize an error

Parameters:

Name Type Description Default
module required
error required
spec_version required

Returns:

Type Description
dict

reload_type_registry(use_remote_preset: bool = True, auto_discover: bool = True)

Reload type registry and preset used to instantiate the SubtrateInterface object. Useful to periodically apply changes in type definitions when a runtime upgrade occurred

Parameters:

Name Type Description Default
use_remote_preset bool True
auto_discover bool True

register_extension(extension: Extension)

Register an Extension and adds its functionality to the ExtensionRegistry

Parameters:

Name Type Description Default
extension Extension required

ExtrinsicReceipt

Object containing information of submitted extrinsic. Block hash where extrinsic is included is required when retrieving triggered events or determine if extrinsic was succesfull

extrinsic_idx: int property

Retrieves the index of this extrinsic in containing block

Returns:

Type Description
int

extrinsic: GenericExtrinsic property

Retrieves the Extrinsic subject of this receipt

Returns:

Type Description
Extrinsic

triggered_events: list property

Gets triggered events for submitted extrinsic. block_hash where extrinsic is included is required, manually set block_hash or use wait_for_inclusion when submitting extrinsic

Returns:

Type Description
list

is_success: bool property

Returns True if ExtrinsicSuccess event is triggered, False in case of ExtrinsicFailed In case of False error_message will contain more details about the error

Returns:

Type Description
bool

error_message: Optional[dict] property

Returns the error message if the extrinsic failed in format e.g.:

{'type': 'System', 'name': 'BadOrigin', 'docs': 'Bad origin'}

Returns:

Type Description
dict

weight: Union[int, dict] property

Contains the actual weight when executing this extrinsic

Returns:

Type Description
int (WeightV1) or dict (WeightV2)

total_fee_amount: int property

Contains the total fee costs deducted when executing this extrinsic. This includes fee for the validator ( (Balances.Deposit event) and the fee deposited for the treasury (Treasury.Deposit event)

Returns:

Type Description
int

__init__(substrate: SubstrateInterface, extrinsic_hash: str = None, block_hash: str = None, block_number: int = None, extrinsic_idx: int = None, finalized = None)

Object containing information of submitted extrinsic. Block hash where extrinsic is included is required when retrieving triggered events or determine if extrinsic was succesfull

Parameters:

Name Type Description Default
substrate SubstrateInterface required
extrinsic_hash str None
block_hash str None
finalized None

get_extrinsic_identifier() -> str

Returns the on-chain identifier for this extrinsic in format "[block_number]-[extrinsic_idx]" e.g. 134324-2

Returns:

Type Description
str

create_from_extrinsic_identifier(substrate: SubstrateInterface, extrinsic_identifier: str) -> ExtrinsicReceipt classmethod

Create an ExtrinsicReceipt with on-chain identifier for this extrinsic in format "[block_number]-[extrinsic_idx]" e.g. 134324-2

Parameters:

Name Type Description Default
substrate SubstrateInterface required
extrinsic_identifier str required

Returns:

Type Description
ExtrinsicReceipt

__get_extrinsic_index(block_extrinsics: list, extrinsic_hash: str) -> int staticmethod

Returns the index of a provided extrinsic