EncodedCall

Documentation for eth_defi.event_reader.multicall_batcher.EncodedCall Python class.

class EncodedCall

Bases: object

Multicall payload, minified implementation.

  • Designed for multiprocessing and historical reads

  • Only carry encoded data, not ABI etc. metadata

  • Contain extra_data which allows route to call results from several calls to one handler class

Example:

convert_to_shares_payload = eth_abi.encode(["uint256"], [share_probe_amount])

share_price_call = EncodedCall.from_keccak_signature(
    address=address,
    signature=Web3.keccak(text="convertToShares(uint256)")[0:4],
    function="convertToShares",
    data=convert_to_shares_payload,
    extra_data=None,
)

Attributes summary

func_name

Store ABI function for debugging purposers

address

Contract address

data

Call ABI-encoded payload

extra_data

Use this to match the reader

first_block_number

First block hint when doing historical multicall reading.

call_id

Running counter call id for debugging purposes

Methods summary

__init__(func_name, address, data, extra_data)

call(web3, block_identifier[, from_, gas, ...])

Return raw results of the call.

call_as_result(web3, block_identifier[, ...])

Perform RPC call and return the result as an EncodedCallResult.

from_contract_call(call[, extra_data, ...])

Create poller call from Web3.py Contract proxy object

from_keccak_signature(address, function, ...)

Create poller call directly from a raw function signature

get_curl_info(block_number)

Get human-readable details for debugging.

get_debug_info()

Get human-readable details for debugging.

is_valid_for_block(block_number)

transact(from_, gas_limit)

Build a transaction payload for this call.
func_name: str

Store ABI function for debugging purposers

address: eth_typing.evm.HexAddress

Contract address

data: bytes

Call ABI-encoded payload

extra_data: dict | None

Use this to match the reader

first_block_number: int | None

First block hint when doing historical multicall reading.

Skip calls for blocks that are earlier than this block number.

call_id: int

Running counter call id for debugging purposes

get_debug_info()

Get human-readable details for debugging.

  • Punch into Tenderly simulator

  • Data contains both function signature and data payload

Return type

str

get_curl_info(block_number)

Get human-readable details for debugging.

  • Punch into Tenderly simulator

  • Data contains both function signature and data payload

Parameters

block_number (int) –

Return type

str

static from_contract_call(call, extra_data=None, first_block_number=None)

Create poller call from Web3.py Contract proxy object

Parameters

  • call (web3.contract.contract.ContractFunction) –

  • extra_data (dict | None) –

  • first_block_number (int | None) –

Return type

eth_defi.event_reader.multicall_batcher.EncodedCall

static from_keccak_signature(address, function, signature, data, extra_data, first_block_number=None, ignore_errors=False, state=None)

Create poller call directly from a raw function signature

Parameters
Return type

eth_defi.event_reader.multicall_batcher.EncodedCall

call(web3, block_identifier, from_='0x0000000000000000000000000000000000000000', gas=None, ignore_error=False, silent_error=False, attempts=3, retry_sleep=30.0)

Return raw results of the call.

Example how to read:

erc_7575_call = EncodedCall.from_keccak_signature(
    address=self.vault_address,
    signature=Web3.keccak(text="share()")[0:4],
    function="share",
    data=b"",
    extra_data=None,
)

result = erc_7575_call.call(self.web3, block_identifier="latest")
share_token_address = convert_uint256_bytes_to_address(result)
Parameters

  • ignore_error – Set to True to inform middleware that it is normal for this call to fail and do not log it as a failed call, or retry it.

  • attempts (int) –

    Use built-in retry mechanism for flaky RPC.

    This works regardless of middleware installed. Set to zero to ignore.

    Cannot be used with ignore_errors.

  • gas (int) –

    Gas limit.

    If not given, use 15M limit except for Mantle use 99M.

  • web3 (web3.main.Web3) –

  • block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, int]) –

Returns

Raw call results as bytes

Raises

ValueError – If the call reverts

Return type

bytes

transact(from_, gas_limit)

Build a transaction payload for this call.

Example:

gas_limit = 15_000_000

# function settleDeposit(uint256 _newTotalAssets) public virtual;
call = EncodedCall.from_keccak_signature(
    address=vault.address,
    function="settleDeposit()",
    signature=Web3.keccak(text="settleDeposit(uint256)")[0:4],
    data=convert_uin256_to_bytes(raw_nav),
    extra_data=None,
)
tx_data = call.transact(
    from_=asset_manager,
    gas_limit=gas_limit,
)
tx_hash = web3.eth.send_transaction(tx_data)
assert_transaction_success_with_explanation(web3, tx_hash)
Parameters
Return type

dict

call_as_result(web3, block_identifier, from_='0x0000000000000000000000000000000000000000', gas=15000000, ignore_error=False)

Perform RPC call and return the result as an EncodedCallResult.

See call() for info.

Parameters

  • gas_limit –

    eth_call RPC gas limit.

    Set to 15M by default, assume to be safe on every chain.

  • web3 (web3.main.Web3) –

  • block_identifier (Union[Literal['latest', 'earliest', 'pending', 'safe', 'finalized'], eth_typing.evm.BlockNumber, eth_typing.evm.Hash32, eth_typing.encoding.HexStr, int]) –

Return type

eth_defi.event_reader.multicall_batcher.EncodedCallResult

__init__(func_name, address, data, extra_data, first_block_number=None, call_id=<factory>, _hash=None)
Parameters
Return type

None