read_events

Documentation for eth_defi.event_reader.reader.read_events function.

read_events(web3, start_block, end_block, events=None, notify=None, chunk_size=100, context=None, extract_timestamps=<function extract_timestamps_json_rpc>, filter=None, reorg_mon=None)[source]

Reads multiple events from the blockchain.

Optimized to read multiple events from test blockchains.

Note

For a much faster event reader check eth_defi.reader.multithread.MultithreadEventReader. This implementation is mostly good with EVM test backends or very small block ranges.

  • Scans chains block by block

  • Returns events as a dict for optimal performance

  • Supports interactive progress bar

  • Reads all the events matching signature - any filtering must be done by the reader

See scripts/read-uniswap-v2-pairs-and-swaps.py for a full example.

Example:

json_rpc_url = os.environ["JSON_RPC_URL"]
web3 = Web3(HTTPProvider(json_rpc_url)

web3.middleware_onion.clear()

# Get contracts
Factory = get_contract(web3, "sushi/UniswapV2Factory.json")

start_block = 1
end_block = web3.eth.block_number

filter = Filter.create_filter(
    factory_address,
    [Factory.events.PairCreated],
)

# Read through all the events, all the chain, using a single threaded slow loop.
# Only suitable for test EVM backends.
pairs = []
log: LogResult
for log in read_events(
    web3,
    start_block,
    end_block,
    filter=filter,
    extract_timestamps=None,
):
    # Signature this
    #
    #  event PairCreated(address indexed token0, address indexed token1, address pair, uint);
    #
    # topic 0 = keccak(event signature)
    # topic 1 = token 0
    # topic 2 = token 1
    # argument 0 = pair
    # argument 1 = pair id
    #
    # log for EthereumTester backend is
    #
    # {'type': 'mined',
    #  'logIndex': 0,
    #  'transactionIndex': 0,
    #  'transactionHash': HexBytes('0x2cf4563f8c275e5b5d7a4e5496bfbaf15cc00d530f15f730ac4a0decbc01d963'),
    #  'blockHash': HexBytes('0x7c0c6363bc8f4eac452a37e45248a720ff09f330117cdfac67640d31d140dc38'),
    #  'blockNumber': 6,
    #  'address': '0xF2E246BB76DF876Cef8b38ae84130F4F55De395b',
    #  'data': HexBytes('0x00000000000000000000000068931307edcb44c3389c507dab8d5d64d242e58f0000000000000000000000000000000000000000000000000000000000000001'),
    #  'topics': [HexBytes('0x0d3648bd0f6ba80134a33ba9275ac585d9d315f0ad8355cddefde31afa28d0e9'),
    #   HexBytes('0x0000000000000000000000002946259e0334f33a064106302415ad3391bed384'),
    #   HexBytes('0x000000000000000000000000b9816fc57977d5a786e654c7cf76767be63b966e')],
    #  'context': None,
    #  'event': web3._utils.datatypes.PairCreated,
    #  'chunk_id': 1,
    #  'timestamp': None}
    #
    arguments = decode_data(log["data"])
    topics = log["topics"]
    token0 = convert_uint256_hex_string_to_address(topics[1])
    token1 = convert_uint256_hex_string_to_address(topics[2])
    pair_address = convert_uint256_bytes_to_address(arguments[0])
    pair_id = convert_int256_bytes_to_int(arguments[1])

    token0_details = fetch_erc20_details(web3, token0)
    token1_details = fetch_erc20_details(web3, token1)
Parameters
Returns

Iterate over LogResult instances for each event matched in the filter.

Return type

Iterable[eth_defi.event_reader.logresult.LogResult]