One way to get around the constant HTTP calling is using websocket which is capable of listening to events on the network without making any API calls.
I will show it in the code bellow how to achieve it. However, before you proceed, please make sure you have installed Python version 3.10 and Web3.py v6 . Although Web3 v6 is a beta version as I am writing this, it is a necessity since the vital upgrades for websockets have been done in v6. Otherwise, you will get the following error:
TypeError: As of 3.10, the loop parameter was removed from Lock() since it is no longer necessary
The code:
import json
from web3 import Web3
from websockets import connect
import asyncio
# Notice that unlike http provider, this time, we are using a websocket provider which can be copied from Infura website
w3 = Web3(Web3.WebsocketProvider("wss://goerli.infura.io/ws/v3/{YOUR_INFURA_API}"))
contract_address = "0x5b6c5f2032C2483251C36DA4EAd2EEe9504694dd"
# Main function that is run asynchronously and independently of the rest of the program
async def get_event():
# Initiates the connection between your dapp and the network
async with connect("wss://goerli.infura.io/ws/v3/815e996eff9c4caa8cfe1349781148b6") as ws:
await ws.send(json.dumps({"id": 1, "method": "eth_subscribe", "params": ["logs", {"address": [contract_address]}]}))
# Wait for the subscription completion.
subscription_response = await ws.recv()
print(f"Subscription response: {subscription_response}")
# Connection has been successful, now it enters a loop where thanks to websocket's method recv(), it waits for an event to be emitted.
# If timeout reaches 60 seconds, it does not affect the connection, the program merely gets back to waiting.
while True:
try:
# Wait for the message in websockets and print the contents.
message = await asyncio.wait_for(ws.recv(), timeout=60)
json_message = json.loads(message)
events_hex = json_message['params']['result']['data']
events_hex_indexed = json_message['params']['result']['topics']
print(f"Json loads: {json_message}")
print(f"Hex events: {events_hex}")
print(f"Hex events, indexed: {events_hex_indexed}")
except asyncio.exceptions.TimeoutError:
pass
if __name__ == "__main__":
loop = asyncio.new_event_loop()
loop.run_until_complete(get_event())
However, even this algorithm is not perfect. Unlike the first algorithm which uses http requests to get events, it returns all data in hex format. And also it returns unindexed events in one long hex string under the key "data" and the indexed events are returned in a lost under the key "topics". In order to decode all the hex into a readable format, we would have to develop a complicated algorithm, which I am not going to show here. Instead I have found a much easier solution.
In the first code I have shown, there are methods log_loop and handle_events which can process the events in way that it returns them in a readable format. However unlike in the first example, where the log_loop method runs nonstop by constantly checking for new entries, here it runs only when the websocket receives a message. And as a result the final code makes an API call via Infura only when event emition takes place.
The code, along with descriptions:
import json
from web3 import Web3
from websockets import connect
import asyncio
# add your blockchain connection information
infura_url = 'https://goerli.infura.io/v3/{YOUR_INFURA_API}'
w3 = Web3(Web3.HTTPProvider(infura_url))
contract_address = Web3.to_checksum_address("0x5b6c5f2032C2483251C36DA4EAd2EEe9504694dd")
contract_abi = '[{"anonymous":false,"inputs":[{"indexed":false,"internalType":"string","name":"message","type":"string"},' \
'{"indexed":true,"internalType":"uint256","name":"date","type":"uint256"},{"indexed":true,"internalType":"address","name":"from","type":"address"},' \
'{"indexed":true,"internalType":"address","name":"to","type":"address"},{"indexed":false,"internalType":"uint256","name":"amount","type":"uint256"},' \
'{"indexed":false,"internalType":"uint256","name":"transaction","type":"uint256"}],"name":"NewTrade","type":"event"},' \
'{"inputs":[{"internalType":"string","name":"_message","type":"string"},{"internalType":"address","name":"to","type":"address"},' \
'{"internalType":"uint256","name":"amount","type":"uint256"}],"name":"Trade","outputs":[],"stateMutability":"payable","type":"function"}]'
contract = w3.eth.contract(address=contract_address, abi=contract_abi)
message_event = contract.events.NewTrade()
block_filter = w3.eth.filter({'fromBlock': 'latest', 'address': contract_address})
def handle_events(event):
receipt = w3.eth.wait_for_transaction_receipt(event['transactionHash'])
result = message_event.process_receipt(receipt)
print(result[0]['args'])
def log_loop(event_filter):
entries = event_filter.get_new_entries()
# When message is successfully received and log_loop method is triggered, "event_filter.get_new_entries()" does not catch the message
# successfully on first attempt and remains empty and skips the for loop. This is why there is the infinite loop bellow, which repeats
# the "event_filter.get_new_entries()" until it catches the message so that it can proceed and decode the transaction receipt and
# get the events
while True:
if len(entries) == 0:
entries = event_filter.get_new_entries()
print(f"Length is Zero!!")
continue
else:
print("Passed")
break
print(f"event_filter_length: {entries}")
for event in entries:
handle_events(event)
print(f"event_filter: {event_filter}, event: {event}")
print("")
# Main function that is run asynchronously and independently of the rest of the program
async def get_event():
# Initiates the connection between your dapp and the network
async with connect("wss://goerli.infura.io/ws/v3/{YOUR_INFURA_API}") as ws:
await ws.send(json.dumps({"id": 1, "method": "eth_subscribe", "params": ["logs", {"address": [f'{contract_address}']}]}))
# Wait for the subscription completion.
subscription_response = await ws.recv()
print(f"Subscription response: {subscription_response}")
while True:
try:
# Wait for the message in websockets and print the contents.
await asyncio.wait_for(ws.recv(), timeout=300)
log_loop(block_filter)
except asyncio.exceptions.TimeoutError:
block_filter = w3.eth.filter({'fromBlock': 'latest', 'address': contract_address})
print('Block filter has been reset')
except ValueError as e:
print(e)
if __name__ == "__main__":
loop = asyncio.new_event_loop()
loop.run_until_complete(get_event())
This is one of my first work that I have done in relation to Python, Web3 and Solidity coding. So I am sorry about any gaps in my explanation, if you find any, please correct me in the comments and I will fix it.
EDIT on 20/02/2023:
I have added some improvements to the code, including that it can download ABI of a designated cotract on it's own and extract the event names from it. Thanks to, that when setting a "message_event" variable it tries multiple event names until it stops getting an IndexError and then it prints the event results.
The code:
import json
import requests
from web3 import Web3
from websockets import connect
import asyncio
import warnings
# add your blockchain connection information
events = []
infura_url = 'https://goerli.infura.io/v3/{YOUR_INFURA_API}'
w3 = Web3(Web3.HTTPProvider(infura_url))
contract_address = Web3.to_checksum_address("0x11cdfC6dAD20C272947F73bc83C70051aF36A363")
etherscan_api = 'YOUR_ETHERSCAN_API'
etherscan_url = f'https://api-goerli.etherscan.io/api?module=contract&action=getabi&address={contract_address}&apikey={etherscan_api}'
# contract ABI extraction via Etherscan API
etherscan_response = requests.get(etherscan_url)
etherscan_content = etherscan_response.json()
contract_abi = etherscan_content.get("result")
contract_abi_dict = json.loads(contract_abi)
print(f"ABI: {contract_abi}")
# Event name extraction
for i, j in enumerate(contract_abi_dict):
if contract_abi_dict[i]['type'] == "event":
events.append(contract_abi_dict[i]['name'])
print(contract_abi_dict[i]['name'])
contract = w3.eth.contract(address=contract_address, abi=contract_abi)
block_filter = w3.eth.filter({'fromBlock': 'latest', 'address': contract_address})
def handle_events(event):
for x, y in enumerate(events):
# warning functions need to be set to ignore UserWarning, otherwise it will pop up in the program every time emited event name missmatches
# the event put into "message_event" variable from "events" list
with warnings.catch_warnings():
warnings.simplefilter("ignore")
# message_event gets tried by all event names until the content matches the event name
message_event = eval(f"contract.events.{events[x]}()")
receipt = w3.eth.wait_for_transaction_receipt(event['transactionHash'])
result = message_event.process_receipt(receipt)
try:
# If the event name picked from events list matches the event name emited. It will successfully store the information in "result"
# variable and print it out.
print(f"Result from event {events[x]}: {result[0]['args']}")
break
except IndexError as e:
# If the event name picked from events list missmatched the event name actually emited, no content will be stored into the "result"
# variable and when attempting to print out the result above, it will get an IndexError exception.
print(e)
def log_loop(event_filter):
entries = event_filter.get_new_entries()
# When message is successfully received and log_loop method is triggered, "event_filter.get_new_entries()" does not catch the message
# successfully on first attempt and remains empty and skips the for loop. This is why there is the infinite loop bellow, which repeats
# the "event_filter.get_new_entries()" until it catches the message so that it can proceed and decode the transaction receipt and
# get the events
while True:
if len(entries) == 0:
entries = event_filter.get_new_entries()
print(f"Length is Zero!!")
continue
else:
print("Passed")
break
# print(f"event_filter_length: {entries}")
for event in entries:
handle_events(event)
# print(f"event_filter: {event_filter}, event: {event}")
print("")
# Main function that is run asynchronously and independently of the rest of the program
async def get_event():
global block_filter
# Initiates the connection between your dapp and the network
async with connect("wss://goerli.infura.io/ws/v3/815e996eff9c4caa8cfe1349781148b6") as ws:
await ws.send(json.dumps({"id": 1, "method": "eth_subscribe", "params": ["logs", {"address": [f'{contract_address}']}]}))
# Wait for the subscription completion.
subscription_response = await ws.recv()
print(f"Subscription response: {subscription_response}")
while True:
try:
# Wait for the message in websockets and print the contents.
await asyncio.wait_for(ws.recv(), timeout=300)
log_loop(block_filter)
except asyncio.exceptions.TimeoutError:
block_filter = w3.eth.filter({'fromBlock': 'latest', 'address': contract_address})
print('Block filter has been reset')
except ValueError as e:
print(e)
if __name__ == "__main__":
loop = asyncio.new_event_loop()
loop.run_until_complete(get_event())
