swarm-swarmscan 0.1.0

Async Python client for the Swarmscan network-indexer API. Parity with the swarmscan-rs Rust crate.

swarmscan-py

Async Python client for the Swarmscan network-indexer API (https://api.swarmscan.io). Async-first (httpx), Python 3.10+, fully type-checked (mypy --strict).

Where the sibling bee-py client talks to a single Bee node, swarmscan-py talks to the network-wide indexer: every node's overlay, reachability and geo location, redistribution (storage-incentives) rounds, and on-chain postage / staking events. It is the foundation for network-level analytics tools (storage-incentives reports, neighborhood population, peer geo-enrichment).

This is the Python port of the swarmscan-rs Rust crate, at full method parity. Response shapes are derived from the Go decode structs in the readsi / neighborhood programs of bee-scripts — the canonical consumers of this API — and validated against the live camelCase wire format.

Install

pip install swarm-swarmscan
# or
uv add swarm-swarmscan

The PyPI distribution name is swarm-swarmscan (matching the swarm-bee naming of the bee-py sibling); the import name is swarmscan:

from swarmscan import Client

Quick start

import asyncio
from swarmscan import Client

async def main() -> None:
    async with Client() as client:  # mainnet; Client.testnet() for testnet
        # Newest page of redistribution rounds.
        page = await client.redistribution_rounds()
        for r in page.rounds:
            print(f"round {r.round_number}: {len(r.events)} events")

        # Per-node detail, including geo location.
        node = await client.node("aa11bb22…")
        country = node.location.country if node.location else None
        print(f"country: {country}")

asyncio.run(main())

The client is an async context manager — async with Client() as client closes the underlying httpx.AsyncClient on exit.

Endpoints

Method	Endpoint	Returns
network_dump()	GET /v1/network/dump	NetworkDump (full node list)
node(overlay)	GET /v1/network/nodes/{overlay}	Node
node_by_eth(eth)	GET /v1/network/nodes/ethereum/{eth}	Node
redistribution_rounds(start)	GET /v1/redistribution/rounds?start=	RedistributionRounds (1 page)
postage_batches_created(start)	GET /v1/events/postage-stamp/batch-created?start=	PostageEvents (1 page)
stake_updated(start)	GET /v1/events/staking/stake-updated?start=	StakeEvents (1 page)

Paging helpers built on top:

• collect_redistribution_rounds(max_rounds) — drain up to N rounds.
• redistribution_rounds_until(until_round) — drain down to a cutoff round.

async with Client() as client:
    recent = await client.collect_redistribution_rounds(50)
    since = await client.redistribution_rounds_until(12000)

Paging

Paging differs by endpoint, matching the live API:

• Redistribution rounds have no server cursor. ?start=N returns rounds with roundNumber <= N (inclusive, descending); to page deeper, request the lowest roundNumber of the previous page minus one. The collect_* / *_until helpers compute this for you and include a no-progress guard against a misbehaving cursor.

• Postage / staking events use a string next cursor in the response; an empty string means no further page:

  async with Client() as client:
      cursor: str | None = None
      while True:
          page = await client.postage_batches_created(cursor)
          for ev in page.events:
              print(ev.data.batch_id)
          if not page.next:
              break
          cursor = page.next
  

Constructors

Constructor	Host
Client()	https://api.swarmscan.io (mainnet)
Client.testnet()	https://api.testnet.swarmscan.io
Client(base_url)	arbitrary (proxy / mock)
Client(base_url, http=...)	arbitrary + caller-provided httpx

All constructors use a shared httpx.AsyncClient connection pool with a 60-second read timeout. Pass your own httpx.AsyncClient via http= to set custom timeouts, auth headers, or to share a pool; the client only closes HTTP clients it owns.

BZZ conversion

Token amounts (rewardAmount, totalAmount, stakeAmount, …) are kept as str (wei/PLUR-scale) to avoid precision loss. swarmscan.to_bzz(s) gives a lossy float BZZ view mirroring bee-scripts' util.ToBZZ (value / 1e16):

from swarmscan import to_bzz

to_bzz("15000000000000000")  # -> 1.5
to_bzz("")                    # -> None

Errors

Every fallible call raises a subclass of SwarmscanError:

• SwarmscanArgumentError — invalid input (empty overlay, bad base URL).
• SwarmscanResponseError — non-2xx response; .status / .is_not_found() let you treat a 404 (unknown overlay / Ethereum address) as a miss.
• SwarmscanJsonError — JSON decode failed.
• SwarmscanTransportError — TCP / TLS / DNS / connection refused.

Stack

• httpx (async) for HTTP, asyncio runtime
• frozen dataclasses for typed responses, mypy --strict clean
• logging for request-level debug events (logger swarmscan.http)

Stability

0.x — breaking changes are allowed in minor bumps while the live-API surface settles.