How to subscribe to events in brownie

Question

I am having trouble subscribing to events with brownie.

My interface is imported with

from brownie import interface

From there, I have access to the external methods but not the events. I have found examples in web3js (here) but not with the python library.

Can you provide useful links or insights about this? Thank you.

Answer 1

You may try to use web3py such as:

from web3 import Web3
web3 = Web3(Web3.WebsocketProvider("WEB_SOCKET_URI")
web3_contract = web3.eth.contract(address="CONTRACT_ADDRESS",
                                  abi="CONTRACT_ABI")
event_filter = web3_contract.events["EVENT_NAME"].createFilter(fromBlock="earliest")
all_events = web3_contract.events["EVENT_NAME"].createFilter(fromBlock="earliest").get_all_entries()

while YOUR_CONDITION:
  # Calling get_new_entries() removes  new entries form the stack, storing new entries for processing
  new_events = event_filter.get_new_entries()
  if "EVENT_NAME" in [dict["event"] for dict in new_events]:
    print("Event fired!!)
    break

Answer 2

Here's an example using brownie running on ethereum mainnet. It subscribes to the Sync event of the UniswapV2Pair contract. This event is emitted by the contract when there has been an update in the reserve balances (when a trade has been made). Of course, make sure you have the https://github.com/Uniswap/v2-core/blob/master/contracts/interfaces/IUniswapV2Pair.sol file in your project's interfaces directory.

import asyncio
from web3 import Web3
from brownie import interface


def main():
    weth_usdc_pair_addr = "0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc"
    contract = interface.IUniswapV2Pair(weth_usdc_pair_addr)
    event_filter = contract.events.Sync.createFilter(fromBlock="latest")
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        asyncio.run(log_loop(event_filter, 1))
    except KeyboardInterrupt:
        print(" Cancelled by user, exiting")
    finally:
        loop.close()


async def log_loop(event_filter, poll_interval):
    while True:
        for Sync in event_filter.get_new_entries():
            handle_event(Sync)
        await asyncio.sleep(poll_interval)


def handle_event(event):
    print(Web3.toJSON(event))