Prerequisite

• The URL to a node that has a WebSocket port opened. This can be either from a provider like infura or through a self-hosted Geth/Nethermind node.

Create a basic subscription

The basic subscription will just print the data as received. It can be tested by doing the following:

>>> from sub3 import RPCMaker, Sub3

#RPCMaker is for easy formatting of JSON-RPC calls
>>> rpc = RPCMaker.new_heads()

>>> sub = Sub3("ws://localhost:8546", rpc)
>>> sub.start()
#Connection to the node
connected

#Successful answer from the JSONRPC request
{"jsonrpc":"2.0","id":"1","result":"0x1aaa6ce63bae0597ceadd723fd05e6db"}

#Starts receiving data
{
 "jsonrpc":"2.0",
 "method":"eth_subscription",
 "params":{"subscription":"0x1aaa6ce63bae0597ceadd723fd05e6db",
 "result":{"parentHash":"0x65906581" [...]}
 }

Sub-classing

The data processing can easily be customized by sub-classing the Sub3 class like so:

from sub3 import Sub3

class NewClient(Sub3):

    async def on_data(self, data):
        # add your own data processing logic

    async def on_closed(self, error):
        # add your processing of `closed` message

    async def on_error(self, error):
        # add your processing of `error` message

rpc = RPCMaker.new_heads()

sub = NewClient("ws://localhost:8546", rpc)

sub.start()