This tutorial will show you how to use Ethereum events to listen for new NFT transactions on the blockchain as they are submitted. For the complete code of the script, feel free to scroll down to the bottom of the article.
Events
Events are essential to the Ethereum blockchain; they allow us to create interfaces that update dynamically as contracts complete executions. By watching the events in a block, we can track both NFT transfers and mints in real-time.
Additionally, we can choose only to track specific NFT transactions or NFT transfers from/to a particular Ethereum address.
Setting Up Our Project
Get started by making sure you have the web3.py and websockets libraries installed on your machine:
pip install web3 websockets
Then, we can import the libraries needed for our project:
from web3 import Web3
import asyncio
import json
from websockets import connect
Connecting to Infura
We’re going to connect to Infura’s websockets endpoint so we can subscribe to any new NFT transactions, as well as Infura’s regular HTTP endpoint so we can use the web3.sha3 function.
We can define the following endpoints:
infura_http_url = 'https://mainnet.infura.io/v3/<YOUR_PROJECT_ID>'
infura_ws_url = 'wss://mainnet.infura.io/ws/v3/<YOUR_PROJECT_ID>'
web3 = Web3(Web3.HTTPProvider(infura_http_url))
Make sure to replace <YOUR_PROJECT_ID> with your actual Infura project ID; you can use the same for both endpoints.
Reading ERC-721 and ERC-1155 Events
NFTs on the Ethereum blockchain typically use ERC-721 or ERC-1155, both of which are standards for deploying Non-Fungible Tokens to the network. Just like ERC-20 is used for fungible tokens, it allows developers to create a standardized smart contract with which other interfaces can easily interact.
While ERC-721 only allows you to create non-fungible tokens, ERC-1155 supports fungible and non-fungible tokens. This ability to support both types of tokens would especially be helpful in games, as a currency could be fungible (like silver or gold), and rare items (like armor or accessories) could be collectibles and thus non-fungible.
Subscribing to Contract Events
The eth.subscribe method allows us to subscribe to events happening on the blockchain. Smart contracts can emit events during execution and are usually used to ‘announce’ that a contract executed a significant action. In our case, it would be the transfer of an NFT, but other examples of events could also include ERC-20 transfers or a swap happening on a decentralized exchange.
There’s a slight difference in the events that these standards emit. ERC-721 transactions emit the following event:
Transfer (address from, address to, uint256 tokenId)
ERC-1155 transactions emit the following event:
TransferSingle (address operator, address from, address to, uint256 id, uint256 value)
We can reflect this in our code as follows:
options721 = {
'topics': [
web3.sha3(text='Transfer(address,address,uint256)').hex()
]}
options1155 = {
'topics': [
web3.sha3(text='TransferSingle(address,address,address,uint256,uint256)').hex()
]}
The Web3.py library doesn’t natively support subscriptions yet. However, using the Python websockets library, we can still utilize Infura’s websockets endpoint to subscribe to transactions and events on the blockchain.
request_721 = {"jsonrpc":"2.0", "id":1, "method":"eth_subscribe", "params":["logs", options721]}
request_1155= {"jsonrpc":"2.0", "id":1, "method":"eth_subscribe", "params":["logs", options115]}
request_string_721 = json.dumps(request_721)
request_string_1155 = json.dumps(request_1155)
Reading ERC-721 Transfers
Create a new async method that will connect to Infura’s websocket endpoint:
async def get_event_721():
async with connect(infura_ws_url) as ws:
await ws.send(request_string_721)
subscription_response = await ws.recv()
print(subscription_response)
Now let’s create a while loop to start listening to new ERC-721 events:
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=60)
event = json.loads(message)
By printing event, we get to see the transaction details, including the topics that contain the ERC-721 transfer details:
{'jsonrpc': '2.0', 'method': 'eth_subscription', 'params': {
'subscription': '0x10ff0f781d4b58859a3ae317724935b2211754557a08', 'result': {
'removed': False, 'logIndex': '0x259', 'transactionIndex': '0x141', 'transactionHash': '0xe1b91c33dcbbf5774436224c8d1152633d3943cecaa1fbd296fa5142c7f75b60', 'blockHash': '0xebe46a4b769e2e633486191d6a9de887ae29a1b078b38112cf22b8e389a57e22', 'blockNumber': '0xe8fe72', 'address': '0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2', 'data': '0x000000000000000000000000000000000000000000000000084e5dce26d2c807', 'topics': ['0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef', '0x0000000000000000000000000d4a11d5eeaac28ec3f61d100daf4d40471f1852', '0x0000000000000000000000005ead5462e7d98308e64bfe3c1d76845e5d2794a1']}}}
To retrieve the transaction’s topics that contain the transaction details, we can navigate to event['params']['result']['topics'], which contain the from address, to address, and token ID. For ERC-721, the event stores information as seen below:
- topics[1] = from address (’ 0x’ followed by 24 zeroes and then the actual address)
- topics[2] = to address (’ 0x’ followed by 24 zeroes and then the actual address)
- topics[3] = token ID (in hexadecimal format)
- address = token’s address
Let’s see how to access these in our while loop:
if len(event['params']['result']['topics']) == 4:
result = event['params']['result']
from_address = '0x' + result['topics'][1][26:]
to_address = '0x' + result['topics'][2][26:]
token_id = int(result['topics'][3], 16)
token_address = result['address']
block = int(result['blockNumber'], 16)
tx_hash = result['transactionHash']
print("New ERC-721 transaction with hash {} found in block {} From: {} To: {} Token Address: {} Token ID: {}".format(tx_hash, block, from_address, to_address, token_address, token_id))
pass
except:
pass
Keeping in mind the rules listed above, we can access the data we are interested in from the event and then print it; the final result looks something like this:
New ERC-721 transaction with hash 0x97f0eda616c51d8e5d28ff8a7d05fdbabf3a1f74c60b698e300d1171b86109eb found in block 15269490 From: 0x617e70c6466499599136470d3a13cd73b4128c84 To: 0xcdd7df995284376fd6af4c86714e828cd3f260b7 Token Address: 0x4050c9b0f9e008e7e8335d1f12bf95e11013fc01 Token ID: 3084
Additionally, if we would like to track transfers made to/from a specific address, we can compare the to_address or from_address from the event with the desired address.
Note: If a transaction contains a new mint, its from parameter will be 0x0000000000000000000000000000000000000000.
Reading ERC-1155 Transfers
For ERC-1155, the process is very similar. We need to take into account that the event has a slightly different distribution of the data within it:
- topics[2] = from address
- topics[3] = to address
- data = spread into two 64-character sections, the first representing the token ID in hexadecimal format and the second section representing the value transferred in hexadecimal format as well
To get the token ID, we’ll need to get the first 66 characters (to account for the ‘0x’ at the beginning plus the next 64 characters) of the data field from the event and then transform it into an integer using the int() function:
token_id = int(result['data'][:66], 16)
Here’s the full function for reading ERC-1155 transfers:
async def get_event_1155():
async with connect(infura_ws_url) as ws:
await ws.send(request_string_1155)
subscription_response = await ws.recv()
print(subscription_response)
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=60)
event = json.loads(message)
result = event['params']['result']
from_address = '0x' + result['topics'][2][26:]
to_address = '0x' + result['topics'][3][26:]
token_id = int(result['data'][:66], 16)
token_address = result['address']
block = int(result['blockNumber'], 16)
tx_hash = result['transactionHash']
print(
"New ERC-1155 transaction with hash {} found in block {} From: {} To: {} Token Address: {} Token ID: {}".format(
tx_hash, block, from_address, to_address, token_address, token_id))
pass
except:
pass
Complete Code Overview
from web3 import Web3
import asyncio
import json
from websockets import connect
infura_http_url = 'https://mainnet.infura.io/v3/<YOUR_PROJECT_KEY>'
infura_ws_url = 'wss://mainnet.infura.io/ws/v3/<YOUR_PROJECT_KEY>'
web3 = Web3(Web3.HTTPProvider(infura_http_url))
options721 = {
'topics': [
web3.sha3(text='Transfer(address,address,uint256)').hex()
]
}
options1155 = {
'topics': [
web3.sha3(text='TransferSingle(address,address,address,uint256,uint256)').hex()
]
}
request_721 = {"jsonrpc":"2.0", "id": 1, "method": "eth_subscribe", "params": ["logs", options721]}
request_1155 = {"jsonrpc":"2.0", "id": 1, "method": "eth_subscribe", "params": ["logs", options1155]}
request_string_721 = json.dumps(request_721)
request_string_1155 = json.dumps(request_1155)
async def get_event_1155():
async with connect(infura_ws_url) as ws:
await ws.send(request_string_1155)
subscription_response = await ws.recv()
print(subscription_response)
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=60)
event = json.loads(message)
result = event['params']['result']
from_address = '0x' + result['topics'][2][26:]
to_address = '0x' + result['topics'][3][26:]
token_id = int(result['data'][:66], 16)
token_address = result['address']
block = int(result['blockNumber'], 16)
tx_hash = result['transactionHash']
print(
"New ERC-1155 transaction with hash {} found in block {} From: {} To: {} Token Address: {} Token ID: {}".format(
tx_hash, block, from_address, to_address, token_address, token_id))
pass
except:
pass
async def get_event_721():
async with connect(infura_ws_url) as ws:
await ws.send(request_string_721)
subscription_response = await ws.recv()
print(subscription_response)
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=60)
event = json.loads(message)
print(event)
if len(event['params']['result']['topics']) == 4:
result = event['params']['result']
from_address = '0x' + result['topics'][1][26:]
to_address = '0x' + result['topics'][2][26:]
token_id = int(result['topics'][3], 16)
token_address = result['address']
block = int(result['blockNumber'], 16)
tx_hash = result['transactionHash']
print("New ERC-721 transaction with hash {} found in block {} From: {} To: {} Token Address: {} Token ID: {}".format(tx_hash, block, from_address, to_address, token_address, token_id))
pass
except:
pass
if __name__ == "__main__":
loop = asyncio.get_event_loop()
while True:
# loop.run_until_complete(get_event_1155())
loop.run_until_complete(get_event_721())
Thanks for sticking to the end, and a special thanks to @wtzb for helping create this article!