Ethereum: How to track and handle events reliably in Python?

Petar Stoykov

June 26, 2024 in Tech, Tutorials

Ensuring data consistency and a seamless user experience is critical when developing on Ethereum. Relying on a single node for event monitoring can be risky due to potential downtimes or connectivity issues.

Event listeners play an essential role in blockchain technology, enabling applications to respond to specific events emitted by smart contracts. These events are vital for decentralized applications (DApps), such as tracking token transfers and monitoring contract executions.

In the Ethereum ecosystem, events are logged on the blockchain and can be listened to by off-chain applications to trigger actions or update their state based on the latest blockchain activity.

Implementing a redundant event listener can solve these problems by subscribing to multiple Ethereum nodes. This approach offers several key benefits:

Increased reliability: Subscribing to multiple nodes minimizes the risk of missing crucial events due to node failures.
Better fault tolerance: The system continues to function effectively even if some nodes experience issues.
Improved performance: Multiple nodes can lower latency and handle higher transaction volumes.

This guide will show you how to build a resilient event listener using Python and web3.py. By leveraging Chainstack’s robust infrastructure, you’ll ensure your application remains reliable and responsive, even in the face of network challenges. Let’s get started!

How to build a redundant Ethereum event listener in Python?

This tutorial aims to build a resilient Ethereum event listener using Python. By leveraging multiple Chainstack endpoints across different regions, we aim to achieve better consistency in catching events and ensure redundancy in case of endpoint failures.

This approach enhances the reliability and robustness of the event listener, making it more effective in real-world scenarios where network issues or endpoint downtimes might occur.

Prerequisites

Ensure you have the following setup:

Python: Make sure Python is installed on your machine.
Virtual environment: Set up and activate a virtual environment. python -m venv venv source venv/bin/activate # On Windows, use `venv\\\\Scripts\\\\activate`
Web3 library: Install the web3.py library using pip. pip install web3
Ethereum node endpoints: Deploy at least three RPC nodes (one Global Node and two Trader Nodes) on Chainstack.

How to set up the redundant event listener Python script?

Create a new file named main.py and paste the following code:

import os
import asyncio
import logging
from web3 import Web3
from web3.middleware import geth_poa_middleware
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
# List of endpoints
endpoints = [
    "ENDPOINT_1",
    "ENDPOINT_2",
    "ENDPOINT_3"
]
# Filter for WETH transfer events
logs_filter = {
    'address': '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2',  # WETH contract address
    'topics': ['0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'],  # Transfer event signature
}
# Set to track seen event identifiers to prevent duplicates
seen_events = set()
async def handle_event(event, endpoint):
    event_id = f"{event['transactionHash'].hex()}-{event['logIndex']}"
    if event_id not in seen_events:
        seen_events.add(event_id)
        logging.info(f"Event received first from {endpoint}: {event_id}")
async def subscribe_to_logs(endpoint):
    while True:
        try:
            web3 = Web3(Web3.HTTPProvider(endpoint))
            web3.middleware_onion.inject(geth_poa_middleware, layer=0)
            if not web3.is_connected():
                logging.warning(f"Failed to connect to endpoint {endpoint}")
                await asyncio.sleep(5)
                continue
            logging.info(f"Connected to endpoint {endpoint}")
            event_filter = web3.eth.filter(logs_filter)
            while True:
                for event in event_filter.get_new_entries():
                    await handle_event(event, endpoint)
                await asyncio.sleep(1)
        except Exception as e:
            logging.error(f"Error when subscribing to logs from {endpoint}: {e}")
            await asyncio.sleep(5)  # Wait before retrying
async def main():
    tasks = [subscribe_to_logs(endpoint) for endpoint in endpoints]
    await asyncio.gather(*tasks)
if __name__ == "__main__":
    asyncio.run(main())

Import libraries: Import os, asyncio, and logging for general purposes, and Web3 and geth_poa_middleware from the web3 library to interact with the Ethereum blockchain.
Configure logging: Set up logging to display informational messages.
Define endpoints: List of Ethereum node endpoints.
Set up event filter: Configure the event filter to listen for WETH transfer events.
Track seen events: Use a set to track event identifiers that have already been processed.
Handle events: Process received events and log new ones.
Subscribe to logs: Continuously connect to an Ethereum node, set up a filter, and process new entries.
Run the main function: Create tasks for each endpoint and run them concurrently using asyncio.gather.

Add your endpoints in the endpoints list and run the following command in the terminal:

python3 main.py

It will start listening and logging events. Example output:

2024-05-16 17:07:09,452 - INFO - Connected to endpoint <https://nd-974-620-518.p2pify.com/>
2024-05-16 17:07:11,345 - INFO - Connected to endpoint <https://nd-777-597-727.p2pify.com/>
2024-05-16 17:07:12,776 - INFO - Connected to endpoint <https://ethereum-mainnet.core.chainstack.com/>
2024-05-16 17:07:13,805 - INFO - Event received first from <https://ethereum-mainnet.core.chainstack.com/:> 0xcab7994c8ff1495136db4966f4ed4556513540c6cf08dbd22e09fb3496acadef-1

Bringing it all together

Creating a resilient Ethereum event listener is essential for ensuring reliable and consistent decentralized applications. By utilizing multiple endpoints across different regions, you can enhance event capture and introduce redundancy, making your event listener more robust against network issues and downtimes.

This tutorial has walked you through the process of setting up, configuring, and running a resilient event listener using Python and web3.py. With this setup, you’ll be well-prepared to handle blockchain events efficiently and reliably, significantly boosting the performance and reliability of your DApps.