Follow a live blockchain from Python with asyncio and 20 lines of code

#python #asyncio #websockets #tutorial

Most asyncio tutorials fake their data source with asyncio.sleep. This one uses a real feed: a public blockchain
network that produces a block every 15 seconds and pushes each one over a WebSocket. There is no API key, no account
and no rate plan, so everything below runs as written.

The network is Janzeer, a small Layer-1 chain. You do not need to know anything about it to follow along; treat it as
a server that emits one JSON event every 15 seconds.

Set up

python -m venv .venv && . .venv/bin/activate
pip install "janzeer-sdk[ws]"

Python 3.10 or newer. The [ws] extra pulls in websockets.

1. Print every new block

import asyncio
from collections import deque

from janzeer import JanzeerRpcWs

NODE = "https://explorer.janzeer.org"


async def main() -> None:
    blocks: asyncio.Queue[dict] = asyncio.Queue()
    gaps: deque[float] = deque(maxlen=20)
    last_ts = None

    async with await JanzeerRpcWs.connect(NODE) as ws:
        sub = await ws.subscribe_new_blocks(blocks.put_nowait)
        try:
            for _ in range(3):
                block = await asyncio.wait_for(blocks.get(), timeout=60)
                ts = block["timestamp"] / 1000
                if last_ts is not None:
                    gaps.append(ts - last_ts)
                last_ts = ts
                avg = sum(gaps) / len(gaps) if gaps else 0.0
                print(f"#{block['height']}  {block['hash'][:12]}…  txs={block['transactionsCount']}  avg gap={avg:.1f}s")
        finally:
            await sub.unsubscribe()


asyncio.run(main())

Output from a real run:

#239213  4e0054fc98db…  txs=1  avg gap=0.0s
#239214  f92c4915328c…  txs=1  avg gap=15.0s
#239215  4e5cf64b2332…  txs=1  avg gap=15.0s

Three things in there are worth copying into your own code.

A queue as the callback sink. subscribe_new_blocks takes a plain callback. Passing blocks.put_nowait turns the
push feed into something you can await, and keeps your processing out of the WebSocket reader. If your handler is
slow, the queue absorbs the burst; the socket keeps being read.

A timeout on every wait. asyncio.wait_for(blocks.get(), timeout=60) is the difference between a script that
reports "the feed went quiet" and one that hangs for ever. Sixty seconds is four block intervals here.

try / finally around the subscription. The async with closes the socket, and the finally unsubscribes
first, so the server is told before the connection drops. On Ctrl-C, asyncio.run cancels main, and both still run.

2. Watch an address

The same feed can filter by address. A validator's wallet is a convenient target because it receives a block reward
every few minutes:

import asyncio

from janzeer import JanzeerRpcWs, format_jnz

NODE = "https://explorer.janzeer.org"
WATCH = ["0x038cba8510852bd5ce07166427d16cf81b6fb849"]


async def main() -> None:
    events: asyncio.Queue[dict] = asyncio.Queue()
    async with await JanzeerRpcWs.connect(NODE) as ws:
        await ws.subscribe_address_activity(WATCH, events.put_nowait)
        event = await asyncio.wait_for(events.get(), timeout=300)
        tx = event["transaction"]
        print(f"block {event['blockHeight']}: {tx['type']} of {format_jnz(tx['amount'])} ({tx['status']})")


asyncio.run(main())

block 239222: reward of 10 JNZ (FINAL)

The status is FINAL in the very first notification. On this network a committed block cannot be reorganised, so
there is no "wait for six confirmations" loop to write. That removes a whole class of state from the client.

Amounts arrive as decimal strings and format_jnz prints them. The SDK never converts money to float; if you do
arithmetic on amounts, use its helpers or decimal.Decimal.

3. Run several things in one loop

Because the clients are ordinary asyncio objects, combining them is asyncio.gather or a TaskGroup. This reads the
chain's current height over HTTP while the WebSocket subscription is open:

import asyncio

from janzeer import AsyncJanzeerRpc, JanzeerRpcWs

NODE = "https://explorer.janzeer.org"


async def main() -> None:
    async with AsyncJanzeerRpc(NODE) as rpc, await JanzeerRpcWs.connect(NODE) as ws:
        blocks: asyncio.Queue[dict] = asyncio.Queue()
        await ws.subscribe_new_blocks(blocks.put_nowait)
        tip, block = await asyncio.gather(rpc.get_tip(), asyncio.wait_for(blocks.get(), 60))
        print("tip over HTTP:", tip["height"], "| next block over WebSocket:", block["height"])


asyncio.run(main())

Both async HTTP clients accept an http= argument, so several clients can share one httpx.AsyncClient and its
connection pool. The SDK's asyncio guide
covers that, and was written by an outside contributor last week.

What happens when the connection drops

The WebSocket client reconnects by itself and resumes a block subscription from the last block it delivered, so a
restart of your Wi-Fi does not lose or repeat blocks. If you need the same guarantee across restarts of your own
process, store the last height you handled and pass it back as from_height when you subscribe; the node replays from
there before following the tip.

Where to go next

Sending a signed transfer and waiting for it to become final is 15 more lines: examples/07_async_quickstart_transfer.py.
The package is on PyPI as janzeer-sdk, typed (py.typed), Apache-2.0.
The live view of the same feed, drawn in a browser: https://explorer.janzeer.org/workflow.

DEV Community