How to listen for Ethereum events using Web3.py in Python

In Python Web3 application development systems use logs to capture what’s going on at a specific moment in time. Applications typically log:

• When the application makes a request to another process or service
• An audit trail to track changes in data
• Exceptions or errors that can impact the stability of the system
• Suspicious activities or attempts at undermining the security of an application
• Events or activities that a user can make in the application

Logs are very useful but sometimes application owners fail to realize that proactive logging also enables enhanced business decisions. Application data fuels business intelligence which opens the possibilities for delivering better performance. Events from the logs are one of the least collected pieces of information. However they are one of the most valuable sources of data for decision making. Do not underestimate the power of system logs.

Ethereum Logs using Web3.py

The Ethereum blockchain logs events that are useful and can be queried. As smart contracts perform actions they can emit events to the logs in the blockchain environment. Click here for a step by step guide on how to add and emit events in your smart contract. System Logs are not part of the actual blockchain since they are not required for consensus.

You can use the Ethereum logs as events for your application. As smart contracts emit events you can have your application subscribe to these events and preform an action when something of interest occurs.

Below is an example of how to subscribe to Ethereum log events using Web3.py in Python. I chose to use the uniswap smart contract as the example in the code below due to the high volume of activity. As the uniswap smart contract emits events about newly created liquidity trading pairs these actions are saved to the log. The Python code below runs in a loop, listens for newly created liquidity trading pair events, and prints the log message to the IDE console.

Before you get started make sure you have:

• An IDE and Python installed
• A connection point (e.g. Infura)
• Web3.py installed in your Python environment

Read the comments in the Python code below to understand how to listen for events on the Ethereum blockchain.

# import the following dependencies
import json
from web3 import Web3
import asyncio

# add your blockchain connection information
infura_url = 'ADDYOURINFURAURL'
web3 = Web3(Web3.HTTPProvider(infura_url))

# uniswap address and abi
uniswap_router = '0x7a250d5630B4cF539739dF2C5dAcb4c659F2488D'
uniswap_factory = '0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f'
uniswap_factory_abi = json.loads('[{"inputs":[{"internalType":"address","name":"_feeToSetter","type":"address"}],"payable":false,"stateMutability":"nonpayable","type":"constructor"},{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"token0","type":"address"},{"indexed":true,"internalType":"address","name":"token1","type":"address"},{"indexed":false,"internalType":"address","name":"pair","type":"address"},{"indexed":false,"internalType":"uint256","name":"","type":"uint256"}],"name":"PairCreated","type":"event"},{"constant":true,"inputs":[{"internalType":"uint256","name":"","type":"uint256"}],"name":"allPairs","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"allPairsLength","outputs":[{"internalType":"uint256","name":"","type":"uint256"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"tokenA","type":"address"},{"internalType":"address","name":"tokenB","type":"address"}],"name":"createPair","outputs":[{"internalType":"address","name":"pair","type":"address"}],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":true,"inputs":[],"name":"feeTo","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[],"name":"feeToSetter","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":true,"inputs":[{"internalType":"address","name":"","type":"address"},{"internalType":"address","name":"","type":"address"}],"name":"getPair","outputs":[{"internalType":"address","name":"","type":"address"}],"payable":false,"stateMutability":"view","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"_feeTo","type":"address"}],"name":"setFeeTo","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"},{"constant":false,"inputs":[{"internalType":"address","name":"_feeToSetter","type":"address"}],"name":"setFeeToSetter","outputs":[],"payable":false,"stateMutability":"nonpayable","type":"function"}]')

contract = web3.eth.contract(address=uniswap_factory, abi=uniswap_factory_abi)


# define function to handle events and print to the console
def handle_event(event):
    print(Web3.toJSON(event))
    # and whatever


# asynchronous defined function to loop
# this loop sets up an event filter and is looking for new entires for the "PairCreated" event
# this loop runs on a poll interval
async def log_loop(event_filter, poll_interval):
    while True:
        for PairCreated in event_filter.get_new_entries():
            handle_event(PairCreated)
        await asyncio.sleep(poll_interval)


# when main is called
# create a filter for the latest block and look for the "PairCreated" event for the uniswap factory contract
# run an async loop
# try to run the log_loop function above every 2 seconds
def main():
    event_filter = contract.events.PairCreated.createFilter(fromBlock='latest')
    #block_filter = web3.eth.filter('latest')
    # tx_filter = web3.eth.filter('pending')
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(
            asyncio.gather(
                log_loop(event_filter, 2)))
                # log_loop(block_filter, 2),
                # log_loop(tx_filter, 2)))
    finally:
        # close loop to free up system resources
        loop.close()


if __name__ == "__main__":
    main()

After running the Python code above the IDE console will print log messages when the PairCreated event is emitted from the uniswap contract. As you can see from the screen shot below the log shows token0 address, token1 address, new pair address, etc.

This log information can be parsed in used in your application to drive other functions. For example:

• Update the user interface
• Query uniswap for a new price
• Place a trade
• etc

Next – Send an ETH transaction using Web3.py in Python

-->