Source code for eth_defi.event

"""Solidity events scalable fetch and reading."""

from typing import Iterable, Optional, Type

from eth_abi.codec import ABICodec
from eth_typing import BlockNumber, HexAddress
from web3 import Web3
from import get_event_data
from web3._utils.filters import construct_event_filter_params
from web3.contract.contract import ContractEvent
from web3.datastructures import AttributeDict

[docs]def fetch_all_events( web3: Web3, event: Type[ContractEvent], address: Optional[HexAddress] = None, argument_filters: Optional[dict] = None, from_block: Optional[BlockNumber] = 1, to_block: Optional[BlockNumber] = None, ) -> Iterable[AttributeDict]: """Get events using eth_getLogs API. This is a stateless method, as oppose to JSON-RPC filter objects. It can be safely called against nodes which do not provide `eth_newFilter` API, like Infura. We are not doing any throttling or API error recovery: If you ask for too many events once this function and your Ethereum node are likely to blow up. Example how to get all ERC-20 transfers to a target address: .. code-block:: python IERC20 = get_contract(web3, "IERC20.json") Transfer = all_transfers_to_user = list(fetch_all_events(web3, Transfer, argument_filters={"to": owner})) :param web3: Web3 instance :param event: Event class grabbed from a Contract proxy class, like ``. :param address: The smart contract address of the event emitter. Set to none to capture events from all the smart contracts. :param argument_filters: Filters based on the event structure, e.g. `to` field `` :param from_block: Limit block range. Set to `1` to get all the events, ever. :param to_block: Limit block range """ # Currently no way to poke this using a public API. # This will return raw underlying ABI JSON object for the event abi = event._get_event_abi() # Depending on the Solidity version used to compile # the contract that uses the ABI, # it might have Solidity ABI encoding v1 or v2. # We just assume the default that you set on Web3 object here. # More information here codec: ABICodec = web3.codec # Here we need to poke a bit into Web3 internals, as this # functionality is not exposed by default. # Construct JSON-RPC raw filter presentation based on human readable Python descriptions # Namely, convert event names to their keccak signatures # More information here: # data_filter_set, event_filter_params = construct_event_filter_params(abi, codec, address=address, argument_filters=argument_filters, fromBlock=from_block, toBlock=to_block) # Call JSON-RPC API on your Ethereum node. # get_logs() returns raw AttributedDict entries logs = web3.eth.get_logs(event_filter_params) # Convert raw binary data to Python proxy objects as described by ABI for log in logs: # Convert raw JSON-RPC log result to human readable event by using ABI data # More information how processLog works here # evt = get_event_data(codec, abi, log) yield evt