Utility functions

Functions:

keccak256	Compute the sha3_256 flavor hash.
resolve	Get value from nested dictionary by dotted path.

Classes:

Web3Client

Implementation of web3 client.

matic.utils.keccak256(list_of_bytes: Iterable[bytes]) → bytes

Compute the sha3_256 flavor hash.

Parameters

list_of_bytes – A list of bytes to be hashed.

Returns

Hash value in bytes (32 bytes).

Return type

bytes

Raises

TypeError – If bytes or bytearray is used instead of sequence as input.

matic.utils.resolve(obj: dict[str, Any], path: Union[str, Iterable[str]]) → Any

Get value from nested dictionary by dotted path.

class matic.utils.Web3Client(provider: web3.providers.base.BaseProvider)

Bases: matic.abstracts.BaseWeb3Client

Implementation of web3 client.

Methods:

read	Perform a reading (non-modifying) operation.
write	Perform a writing (modifying) operation.
get_contract	Obtain a contract from deployment address and ABI dictionary.
estimate_gas	Estimate gas amount for transaction.
get_transaction_count	Get amount of transactions in specified block for given address.
get_transaction	Obtain transaction object by hash.
get_transaction_receipt	Get receipt for transaction.
get_block	Get block (with raw transaction data) by hash or number.
get_block_with_transaction	Get block (with decoded transaction data) by hash or number.
send_rpc_request	Perform arbitrary RPC request.
encode_parameters	Encode ABI parameters according to schema.
decode_parameters	Decode binary data to ABI parameters according to schema.
etherium_sha3	Calculate solidity keccak hash of given values (encoded as types).

Attributes:

gas_price	Current gas price.
chain_id	Current chain id.

read(config: matic.json_types.ITransactionRequestConfig, return_transaction: bool = False) → Any

Perform a reading (non-modifying) operation.

write(config: matic.json_types.ITransactionRequestConfig, private_key: Optional[str] = None, return_transaction: bool = False) → matic.web3_client.TransactionWriteResult

Perform a writing (modifying) operation.

get_contract(address: eth_typing.evm.HexAddress, abi: dict[str, Any]) → matic.web3_client.Web3Contract

Obtain a contract from deployment address and ABI dictionary.

property gas_price: int

Current gas price.

estimate_gas(transaction: matic.json_types.ITransactionRequestConfig) → int

Estimate gas amount for transaction.

get_transaction_count(address: str, block_number: int) → int

Get amount of transactions in specified block for given address.

property chain_id: int

Current chain id.

get_transaction(transaction_hash: bytes) → matic.json_types.ITransactionData

Obtain transaction object by hash.

get_transaction_receipt(transaction_hash: bytes, timeout: int = 120) → matic.json_types.ITransactionReceipt

Get receipt for transaction.

get_block(block_hash_or_block_number: Union[Literal['latest', 'earliest', 'pending'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, hexbytes.main.HexBytes, int]) → matic.json_types.IBlock

Get block (with raw transaction data) by hash or number.

get_block_with_transaction(block_hash_or_block_number: Union[Literal['latest', 'earliest', 'pending'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, hexbytes.main.HexBytes, int]) → matic.json_types.IBlockWithTransaction

Get block (with decoded transaction data) by hash or number.

send_rpc_request(method: web3.types.RPCEndpoint, params: Iterable[Any]) → web3.types.RPCResponse

Perform arbitrary RPC request.

encode_parameters(params: Sequence[Any], types: Sequence[str]) → bytes

Encode ABI parameters according to schema.

decode_parameters(binary_data: bytes, types: Sequence[Any]) → tuple[Any, ...]

Decode binary data to ABI parameters according to schema.

etherium_sha3(types: Iterable[str], values: Iterable[Any]) → bytes

Calculate solidity keccak hash of given values (encoded as types).

ABI manager

Classes:

ABIManager

Caching manager for fetched contract ABI dicts.

class matic.utils.abi_manager.ABIManager(network_name: str, version: str)

Bases: object

Caching manager for fetched contract ABI dicts.

Methods:

get_config	Get option from cache item by dotted path.
get_abi	Get ABI dict for contract and memoise it.
set_abi	Store ABI dict in cache.

get_config(path: str) → Any

Get option from cache item by dotted path.

get_abi(contract_name: str, bridge_type: Optional[str] = None) → dict[str, Any]

Get ABI dict for contract and memoise it.

set_abi(contract_name: str, bridge_type: str, abi: dict[str, Any]) → None

Store ABI dict in cache.

Proof generation

Functions:

removeprefix	Return a str with the given prefix string removed if present.
get_fast_merkle_proof	Get fast Merkle proof for block.
build_block_proof	Get proof for block as single byte string.
query_root_hash	Get root hash.
recursive_zero_hash	Get n-th recursive zero hash.
get_receipt_proof	Get proof for receipt.
is_typed_receipt	Check if transaction was performed and type is non-zero.
get_state_sync_tx_hash	Get block's tx hash for state-sync receipt.
get_receipt_bytes	Get binary representation of receipt for storing in trie.

matic.utils.proof_utils.removeprefix(self, prefix, /)

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

matic.utils.proof_utils.get_fast_merkle_proof(web3: matic.abstracts.BaseWeb3Client, block_number: int, start_block: int, end_block: int) → list[bytes]

Get fast Merkle proof for block.

matic.utils.proof_utils.build_block_proof(matic_web3: matic.abstracts.BaseWeb3Client, start_block: int, end_block: int, block_number: int) → bytes

Get proof for block as single byte string.

matic.utils.proof_utils.query_root_hash(client: matic.abstracts.BaseWeb3Client, start_block: int, end_block: int) → bytes

Get root hash.

matic.utils.proof_utils.recursive_zero_hash(n: int, client: matic.abstracts.BaseWeb3Client) → bytes

Get n-th recursive zero hash.

matic.utils.proof_utils.get_receipt_proof(receipt: matic.json_types.ITransactionReceipt, block: matic.json_types.IBlockWithTransaction, web3: matic.abstracts.BaseWeb3Client, request_concurrency: Optional[int] = None, receipts_val: Optional[Iterable[matic.json_types.ITransactionReceipt]] = None) → matic.json_types.IReceiptProof

Get proof for receipt.

matic.utils.proof_utils.is_typed_receipt(receipt: matic.json_types.ITransactionReceipt) → bool

Check if transaction was performed and type is non-zero.

matic.utils.proof_utils.get_state_sync_tx_hash(block: matic.json_types.IBaseBlock) → bytes

Get block’s tx hash for state-sync receipt.

Bor blockchain includes extra receipt/tx for state-sync logs, but it is not included in transactionRoot or receiptRoot. So, while calculating proof, we have to exclude them.

This is derived from block’s hash and int.

matic.utils.proof_utils.get_receipt_bytes(receipt: matic.json_types.ITransactionReceipt) → bytes

Get binary representation of receipt for storing in trie.

Web3 side chain client

Classes:

Web3SideChainClient

Web3 client class for a side chain.

class matic.utils.web3_side_chain_client.Web3SideChainClient(config: matic.utils.web3_side_chain_client._C)

Bases: Generic[matic.utils.web3_side_chain_client._C]

Web3 client class for a side chain.

Attributes:

parent	Parent client instance.
child	Child client instance.
abi_manager	ABI manager instance.
main_plasma_contracts	Parameters of plasma contracts.
main_pos_contracts	Parameters of POS contracts.

Methods:

get_abi	Get ABI dictionary for given name and type.
get_config	Fetch configuration option by a dotted path.
is_eip_1559_supported	Check if EIP-1559 (improved fee specification) is available for chain.

parent: matic.abstracts.BaseWeb3Client

Parent client instance.

child: matic.abstracts.BaseWeb3Client

Child client instance.

abi_manager: matic.utils.abi_manager.ABIManager

ABI manager instance.

get_abi(name: str, type_: Optional[str] = None) → dict[str, Any]

Get ABI dictionary for given name and type.

get_config(path: str) → Any

Fetch configuration option by a dotted path.

property main_plasma_contracts: dict[str, Any]

Parameters of plasma contracts.

property main_pos_contracts: dict[str, Any]

Parameters of POS contracts.

is_eip_1559_supported(is_parent: bool) → bool

Check if EIP-1559 (improved fee specification) is available for chain.

Root chain

Classes:

RootChain

Root chain implementation.

class matic.utils.root_chain.RootChain(client: matic.utils.web3_side_chain_client.Web3SideChainClient[matic.utils.root_chain._C], address: eth_typing.evm.HexAddress)

Bases: matic.utils.base_token.BaseToken[matic.utils.root_chain._C]

Root chain implementation.

This represents a connection between parent (root) and child chains. For example, Goerli testnet is a root chain for Mumbai testnet.

Attributes:

last_child_block

Get last block number on child chain.

Methods:

find_root_block_from_child

Find root block corresponding to child block of given number.

property last_child_block: int

Get last block number on child chain.

find_root_block_from_child(child_block_number: int) → int

Find root block corresponding to child block of given number.

Exit data building

Classes:

ExitUtil

Helper utility class for building and performing exit actions with POS bridge.

class matic.utils.exit_util.ExitUtil(client: matic.utils.web3_side_chain_client.Web3SideChainClient[matic.utils.exit_util._C], root_chain: matic.utils.root_chain.RootChain[matic.utils.exit_util._C])

Bases: Generic[matic.utils.exit_util._C]

Helper utility class for building and performing exit actions with POS bridge.

Attributes:

root_chain	Root chain address.
config	Configuration (same as of client).

Methods:

get_chain_block_info	Obtain information about block that includes given transaction.
is_checkpointed	Check if given transaction is checkpointed.
build_payload_for_exit	Build exit payload for transaction hash.
build_multiple_payloads_for_exit	Build exit payload for multiple indices.
get_exit_hash	Build exit hash for burn transaction.

root_chain: matic.utils.root_chain.RootChain[matic.utils.exit_util._C]

Root chain address.

config: matic.utils.exit_util._C

Configuration (same as of client).

get_chain_block_info(burn_tx_hash: bytes) → matic.utils.exit_util._IChainBlockInfo

Obtain information about block that includes given transaction.

is_checkpointed(burn_tx_hash: bytes) → bool

Check if given transaction is checkpointed.

build_payload_for_exit(burn_tx_hash: bytes, index: int, log_event_sig: bytes, is_fast: bool) → bytes

Build exit payload for transaction hash.

build_multiple_payloads_for_exit(burn_tx_hash: bytes, log_event_sig: bytes, is_fast: bool) → list[bytes]

Build exit payload for multiple indices.

get_exit_hash(burn_tx_hash: bytes, index: int, log_event_sig: bytes) → bytes

Build exit hash for burn transaction.

Implementation details

matic.utils.base_token

Classes:

BaseToken

Base class for all tokens.

class matic.utils.base_token.BaseToken(address: eth_typing.evm.HexAddress, is_parent: bool, name: str, client: matic.utils.web3_side_chain_client.Web3SideChainClient[matic.utils.base_token._C], bridge_type: Optional[str] = None)

Bases: Generic[matic.utils.base_token._C]

Base class for all tokens.

Attributes:

contract

Contract object.

Methods:

method	Call arbitrary JSON-RPC method.
process_write	Perform write (modifying) operation.
send_transaction	Sign and send the transaction.
read_transaction	Send read (non-modifying) transaction without RPC method.
process_read	Perform read (non-modifying) operation.
get_client	Get web3 client instance.
create_transaction_config	Fill in missing fields in transaction request.
transfer_erc_20	Transfer ERC-20 token.
transfer_erc_721	Transfer ERC-721 token.
check_for_root	Assert that method is called on parent chain instance.
check_for_child	Assert that method is called on child chain instance.
transfer_erc_1155	Transfer ERC-1155 token.

property contract: matic.abstracts.BaseContract

Contract object.

method(method_name: str, *args: Any) → Any

Call arbitrary JSON-RPC method.

process_write(method: matic.abstracts.BaseContractMethod, option: Optional[matic.json_types.ITransactionOption] = None, private_key: Optional[str] = None) → matic.json_types.ITransactionWriteResult

Perform write (modifying) operation.

Parameters

• method – Method instance (with arguments passed on instantiation)

• option – Additional parameters. May contain special key return_transaction (see Returns section)

• private_key – Sender private key (may be missing, if your provider supports implicit tx signing)

Returns

ITransactionWriteResult if actual write was performed;

ITransactionRequestConfig if return_transaction=True. (builds the final transaction dictionary and returns it).

send_transaction(option: Optional[matic.json_types.ITransactionOption] = None, private_key: Optional[str] = None) → matic.json_types.ITransactionWriteResult

Sign and send the transaction.

read_transaction(option: Optional[matic.json_types.ITransactionOption] = None) → Any

Send read (non-modifying) transaction without RPC method.

Parameters

option – Additional parameters. May contain special key return_transaction (see Returns section)

Returns

ITransactionWriteResult if actual write was performed;

ITransactionRequestConfig if return_transaction=True. (builds the final transaction dictionary and returns it).

process_read(method: matic.abstracts.BaseContractMethod, option: Optional[matic.json_types.ITransactionOption] = None) → Any

Perform read (non-modifying) operation.

Parameters

• method – Method instance (with arguments passed on instantiation)

• option – Additional parameters. May contain special key return_transaction (see Returns section)

Returns

ITransactionWriteResult if actual write was performed;

ITransactionRequestConfig if return_transaction=True. (builds the final transaction dictionary and returns it).

get_client(is_parent: bool) → matic.abstracts.BaseWeb3Client

Get web3 client instance.

create_transaction_config(tx_config: matic.json_types.ITransactionRequestConfig | None, method: matic.abstracts.BaseContractMethod | None, is_parent: bool, is_write: bool) → matic.json_types.ITransactionRequestConfig

Fill in missing fields in transaction request.

Warning: this method may raise if your transaction cannot be executed,

pass gas_limit to prevent it from happening.

transfer_erc_20(to: str, amount: int, private_key: Optional[str] = None, option: Optional[matic.json_types.ITransactionOption] = None)

Transfer ERC-20 token.

transfer_erc_721(from_: str, to: str, token_id: int, private_key: Optional[str] = None, option: Optional[matic.json_types.ITransactionOption] = None)

Transfer ERC-721 token.

check_for_root() → None

Assert that method is called on parent chain instance.

check_for_child() → None

Assert that method is called on child chain instance.

transfer_erc_1155(from_: str, to: str, amount: int, token_id: int, data: bytes | None = b'', private_key: Optional[str] = None, option: Optional[matic.json_types.ITransactionOption] = None)

Transfer ERC-1155 token.

matic.utils.bridge_client

Classes:

BridgeClient

Base class for POS bridge with reusable methods.

class matic.utils.bridge_client.BridgeClient(config: matic.utils.bridge_client._C)

Bases: Generic[matic.utils.bridge_client._C]

Base class for POS bridge with reusable methods.

Attributes:

exit_util	Helper class for exit data building.
client	Actual connecting client.

Methods:

is_checkpointed	Check if transaction is checkpointed.
is_deposited	Check if deposit has finished after exit (StateSynced happened).

exit_util: matic.utils.exit_util.ExitUtil[matic.utils.bridge_client._C]

Helper class for exit data building.

Should be set after instantiation to prevent cycles.

client: matic.utils.web3_side_chain_client.Web3SideChainClient[matic.utils.bridge_client._C]

Actual connecting client.

is_checkpointed(tx_hash: bytes) → bool

Check if transaction is checkpointed.

is_deposited(deposit_tx_hash: bytes) → bool

Check if deposit has finished after exit (StateSynced happened).

matic.utils.merkle_tree

Classes:

MerkleTree

Hash tree.

class matic.utils.merkle_tree.MerkleTree(leaves: Optional[Sequence[bytes]] = None)

Bases: object

Hash tree.

See this article.

Attributes:

leaves	Tree leaves.
layers	Tree layers.
depth	Tree depth.
root	Tree root.

Methods:

estimate_depth	Estimate depth of a tree with given size.
create_hashes	Add nodes to tree.
get_proof	Get proof for leaf as a sequence of nodes.
verify	Verify given proof to match the tree.

leaves: list[bytes]

Tree leaves.

layers: list[list[bytes]]

Tree layers.

static estimate_depth(size: int) → int

Estimate depth of a tree with given size.

Like a heap, hash tree has height equal to log2(size).

property depth: int

Tree depth.

create_hashes(nodes: Sequence[bytes]) → None

Add nodes to tree.

property root: bytes

Tree root.

get_proof(leaf: bytes) → list[bytes]

Get proof for leaf as a sequence of nodes.

verify(value: bytes, index: int, root: bytes, proof: Iterable[bytes]) → bool

Verify given proof to match the tree.

matic.utils.polyfill

Functions:

removeprefix	Return a str with the given prefix string removed if present.
removesuffix	Return a str with the given suffix string removed if present.

matic.utils.polyfill.removeprefix(self, prefix, /)

Return a str with the given prefix string removed if present.

If the string starts with the prefix string, return string[len(prefix):]. Otherwise, return a copy of the original string.

matic.utils.polyfill.removesuffix(self, suffix, /)

Return a str with the given suffix string removed if present.

If the string ends with the suffix string and that suffix is not empty, return string[:-len(suffix)]. Otherwise, return a copy of the original string.