healchain-sdk 0.4.2

Python SDK for HealChain — self-healing decentralized storage via Reed-Solomon erasure coding

healchain-sdk

Python SDK for HealChain — self-healing decentralized storage using Reed-Solomon erasure coding on Ethereum and Arbitrum.

Zero dependencies. Works with Python 3.8+.

Install

pip install healchain-sdk

Quick Start

from healchain import HealChain

hc = HealChain(api_url="https://api.healchain.org")

# Store data — blocks until oracle fulfills
result = hc.store("Hello World", label="my first record")
print(f"Stored as record #{result['record_id']} on chain {result['chain_id']}")

# Retrieve — auto-discovers chain from GlobalRegistry
record = hc.retrieve(result["record_id"])
print(record["text"])  # Hello World

API

HealChain(api_url, **options)

Parameter	Type	Default	Description
api_url	str	https://api.healchain.org	API base URL
api_key	str	None	Optional API key
network	str	'sepolia'	Default network: 'sepolia' or 'arbitrum'
poll_interval	float	4.0	Seconds between oracle fulfillment polls
poll_timeout	float	120.0	Max seconds to wait for fulfillment
data_shards	int	10	Default RS data shards
parity_shards	int	4	Default RS parity shards
timeout	float	30.0	HTTP request timeout in seconds

---

hc.store(data, *, label, network, on_pending, on_fulfilled)

Store data on-chain. Blocks until the oracle fulfills the request.

Parameters:

• data — str or bytes
• label — record label (default: 'sdk upload')
• network — override network: 'sepolia' or 'arbitrum'
• on_pending(info) — callback when tx is submitted
• on_fulfilled(result) — callback when oracle fulfills

Returns:

{
    "record_id":    str,   # on-chain record ID
    "request_id":   str,   # oracle request ID
    "tx":           str,   # submission transaction hash
    "chain_id":     str,   # chain where data is stored
    "original_size": str,  # original size in bytes
    "encoded_size":  str,  # encoded shard size in bytes
}

Example with progress callbacks:

def on_pending(info):
    print(f"Submitted! request_id={info['request_id']}")

def on_fulfilled(result):
    print(f"Fulfilled! record_id={result['record_id']}")

result = hc.store(
    "important data",
    label="archive",
    network="arbitrum",      # ~75x cheaper than Sepolia
    on_pending=on_pending,
    on_fulfilled=on_fulfilled,
)

Storing bytes:

with open("document.pdf", "rb") as f:
    result = hc.store(f.read(), label="my document")

---

hc.retrieve(record_id)

Retrieve a record by ID. Automatically queries the GlobalRegistry to find which chain holds the data.

Returns:

{
    "record_id": str,
    "data":      str,   # hex-encoded original data (0x...)
    "text":      str,   # UTF-8 decoded text
    "bytes":     int,   # original data size
    "chain_id":  str,   # '11155111' (Sepolia) or '421614' (Arbitrum)
}

---

hc.get_metadata(record_id)

Fetch record metadata without retrieving the full data.

Returns: dict with label, owner, original_size, encoded_size, data_shards, parity_shards, timestamp, data_hash

---

hc.list(page=0, limit=10)

List records with pagination.

Returns:

{
    "records": list,   # record summaries
    "total":   int,    # total record count
    "pages":   int,    # total page count
    "page":    int,    # current page
    "limit":   int,    # records per page
}

---

hc.health()

Check service health.

Returns: dict with status, version, geth, lastBlock

---

Error Handling

from healchain import HealChain, HealChainError

hc = HealChain()

try:
    record = hc.retrieve(999999)
except HealChainError as e:
    print(e)           # human-readable message
    print(e.status)    # HTTP status code (int or None)
    print(e.code)      # 'NETWORK_ERROR' | 'FULFILLMENT_TIMEOUT' | None
    print(e.response)  # raw response body (dict or None)

---

Networks

Network	Chain ID	Notes
sepolia	11155111	Ethereum Sepolia testnet
arbitrum	421614	Arbitrum Sepolia testnet — ~75x cheaper gas

Retrieve works automatically regardless of which chain the data is on.

---

Async Usage

The SDK is synchronous by default. For async contexts use a thread executor:

import asyncio
from concurrent.futures import ThreadPoolExecutor
from healchain import HealChain

hc = HealChain()
executor = ThreadPoolExecutor()

async def store_async(data, label):
    loop = asyncio.get_event_loop()
    return await loop.run_in_executor(executor, lambda: hc.store(data, label=label))

result = asyncio.run(store_async("Hello", label="async test"))

---

Async Usage (v0.2+)

Import AsyncHealChain for full async/await support:

import asyncio
from healchain import AsyncHealChain

async def main():
    async with AsyncHealChain(api_url="https://api.healchain.org") as hc:
        result = await hc.store("Hello World", label="my record")
        record = await hc.retrieve(result["record_id"])
        print(record["text"])

asyncio.run(main())

Concurrent operations

async with AsyncHealChain() as hc:
    # Store multiple records concurrently
    results = await hc.store_many([
        ("Hello", "record 1"),
        ("World", "record 2"),
    ], concurrency=3)

    # Retrieve multiple records concurrently
    records = await hc.retrieve_many([0, 1, 2, 3, 4], concurrency=5)

FastAPI integration

from fastapi import FastAPI
from healchain import AsyncHealChain

app = FastAPI()
hc  = AsyncHealChain(api_url="https://api.healchain.org")

@app.get("/retrieve/{record_id}")
async def retrieve(record_id: int):
    return await hc.retrieve(record_id)

Roadmap

• v0.1 — REST API client ✅
• v0.2 — Async support + concurrent store_many/retrieve_many ✅
• v0.3 — Streaming large file support
• v1.0 — Mainnet support

---

License

MIT

---

Built on HealChain · GitHub