0G Storage Python SDK

This repository provides a Python SDK for interacting with 0G Storage.

Installation

• Requires Python 3.9+
• Install (editable): pip install -e .
• Dev deps: pip install -e .[test]

Quick Start (Upload: Indexer or Node)

import os

from zg_storage import EvmClient
from zg_storage.indexer import IndexerClient
from zg_storage.transfer import NodeUploader, NodeUploaderConfig, UploadOption

USE_INDEXER = True

private_key = os.environ["PRIVATE_KEY"]
evm = EvmClient(
    rpc_url="https://evmrpc-testnet.0g.ai",
    private_key=private_key,
)

if USE_INDEXER:
    indexer = IndexerClient(
        "https://indexer-storage-testnet-turbo.0g.ai",
        evm_client=evm,
        flow_address="0x...",
    )
    tx_hash, root_hex = indexer.upload(file_path="your_file.dat", tags=b"", option=UploadOption())
else:
    cfg = NodeUploaderConfig(
        nodes=["http://node-1:5678", "http://node-2:5678"],
        evm_client=evm,
        flow_address="0x...",
    )
    uploader = NodeUploader.from_config(cfg)
    tx_hash, root_hex = uploader.upload(file_path="your_file.dat", tags=b"", option=UploadOption())

print("tx:", tx_hash)
print("root:", root_hex)

Download (Indexer or Node List)

from zg_storage.indexer import IndexerClient
from zg_storage.transfer import DownloadOption, NodeDownloader, NodeDownloaderConfig

USE_INDEXER = True
root = "0x..."
target = "downloaded.bin"

opt = DownloadOption()

if USE_INDEXER:
    indexer = IndexerClient("https://indexer-storage-testnet-turbo.0g.ai")
    indexer.download(root=root, filename=target, option=opt)
else:
    cfg = NodeDownloaderConfig(
        nodes=["http://node-1:5678", "http://node-2:5678"],
    )
    downloader = NodeDownloader.from_config(cfg)
    downloader.download(root=root, target=target, option=opt)

For all detailed parameters, advanced options, and latest endpoint/config guidance, see https://doc.0g.ai.

KV Write (Memory Upload)

import os

from zg_storage import EvmClient
from zg_storage.indexer import IndexerClient
from zg_storage.kv import StreamDataBuilder
from zg_storage.kv.types import create_tags
from zg_storage.core.data import BytesDataSource
from zg_storage.transfer import UploadOption

private_key = os.environ["PRIVATE_KEY"]
evm = EvmClient(rpc_url="https://evmrpc-testnet.0g.ai", private_key=private_key)

indexer = IndexerClient(
    "https://indexer-storage-testnet-turbo.0g.ai",
    evm_client=evm,
    flow_address="0x...",
)

builder = StreamDataBuilder()
builder.set(
    stream_id="0x" + "00" * 32,  # dummy stream id
    key=b"demo_key",             # dummy key
    data=b"demo_value",          # dummy value
)
stream_data = builder.build(sorted_items=True)
payload = stream_data.encode()
tags = create_tags(builder.stream_ids(), sorted_ids=True)

opt = UploadOption(tags=tags)
tx_hash, root_hex = indexer.upload(
    file_path=BytesDataSource(payload),
    tags=tags,
    option=opt,
)

KV Read

from zg_storage.kv import KvClient

kv = KvClient("http://your-kv-node:6789")
data = kv.get_values(
    "0x" + "00" * 32,  # dummy stream id
    ["demo_key"],      # dummy keys
)
print(data)

Development

• Requires Python 3.9+
• Install dev deps: pip install -e .[test]
• Run tests: pytest