3

I am looking for a subscribe function in web3py like the one implemented in web3js:

web3.eth.subscribe('newBlockHeaders' [, callback]);

As there currently is no implementation of it in web3py, has anyone an idea on a possible implementation or some resources on it?

2 Answers 2

11

web3.py doesn't have a native implementation of watchers like web3.js. your only option is to connect to the exposed websocket like follows:

import asyncio
import json
import requests
from websockets import connect

async def get_event():
    async with connect("ws://localhost:8545") as ws:
        await ws.send({"id": 1, "method": "eth_subscribe", "params": ["newHeads"]})
        subscription_response = await ws.recv()
        print(subscription_response)
        # you are now subscribed to the event 
        # you keep trying to listen to new events (similar idea to longPolling)
        while True:
            try:
                message = await asyncio.wait_for(ws.recv(), timeout=60)
                print(json.loads(message))
                pass
            except:
                pass
if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    while True:
        loop.run_until_complete(get_event())

Note that the subscription_response will have a subscription id that you can use to cancel the subscription. you can find all jsonRPC methods form the docs

1
  • For BSC await ws.send('{"id":1,"method":"eth_subscribe","params":["newHeads"]}')
    – Gianmar
    Feb 21, 2022 at 16:56
0

The most straightforward solution is to use filters. You can find several filter usage examples in web3.py docs. Here is the synchronous example:

from web3.auto import w3
import time

def handle_event(event):
    print(event)

def log_loop(event_filter, poll_interval):
    while True:
        for event in event_filter.get_new_entries():
            handle_event(event)
        time.sleep(poll_interval)

def main():
    block_filter = w3.eth.filter('latest')
    log_loop(block_filter, 2)

if __name__ == '__main__':
    main()

Filtering "latest" will return the most recent block hashes, as soon as they become available.

There are also asynchronous examples in the above link.

1
  • 5
    that's not a subscription Jul 16, 2022 at 10:38

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service and acknowledge you have read our privacy policy.

Not the answer you're looking for? Browse other questions tagged or ask your own question.