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: Optional[dict]: Use this to match the reader

first_block_number: Optional[int]

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 (Optional[dict]) –
first_block_number (Optional[int]) –

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

address (eth_typing.evm.HexAddress) –
function (str) –
signature (bytes) –
data (bytes) –
extra_data (Optional[dict]) –
first_block_number (Optional[int]) –
ignore_errors (bool) –
state (Optional[eth_defi.event_reader.multicall_batcher.BatchCallState]) –

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, retry_exceptions=None)

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 – Mark an expected call failure, such as probing an optional Solidity method. This does not suppress the exception; it disables retries unless retry_exceptions explicitly allows one.
attempts (int) –
Use built-in retry mechanism for flaky RPC.

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

With FallbackProvider, this caps its retries after the initial request instead of adding an outer retry loop.
retry_exceptions (Optional[set[type[Exception]]]) – Transient exception classes to retry when ignore_error is set. Other exceptions, including Solidity reverts, are raised after one attempt. With FallbackProvider, the allow-list also lets the provider switch endpoint for the selected failure.
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

from_ (eth_typing.evm.HexAddress) –
gas_limit (int) –

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.

Performs an RPC call and returns a wrapped result in 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

func_name (str) –
address (eth_typing.evm.HexAddress) –
data (bytes) –
extra_data (Optional[dict]) –
first_block_number (Optional[int]) –
call_id (int) –
_hash (int) –

Return type

None