Using Walrus with Python

Using an agent? Try this prompt

Show me how to store and read Walrus blobs from Python using the HTTP API. Include examples for both the REST endpoint and the JSON API, and show how to track blob events.

Walrus does not ship an official Python SDK. Python reaches Walrus through the interfaces that are language-agnostic instead: the HTTP API, the CLI, and Sui JSON-RPC. The following examples show all three: storing and reading blobs with plain requests calls, driving the Walrus CLI's JSON mode as a subprocess, and querying Sui JSON-RPC to track Walrus storage events.

Prerequisites

The complete, runnable example files live in the examples/python directory of the Walrus repository.

Prerequisites

• The walrus CLI and a funded Sui wallet. See Getting Started.

• The requests HTTP library: pip install requests. It is the only third-party dependency.

• Update the constants in utils.py to match your system: PATH_TO_WALRUS (the path to your walrus binary), PATH_TO_WALRUS_CONFIG (the path to your client configuration file), and FULL_NODE_URL (a Sui full node RPC URL, for example https://fullnode.testnet.sui.io:443 for Testnet).

How to use the HTTP API

The HTTP API is the lightest integration path: you store blobs with PUT requests to a publisher and read them with GET requests to an aggregator, with no Walrus-specific dependencies in your Python code. The following example talks to a local walrus daemon, which combines the publisher and aggregator roles on a single address. Start the daemon first, pointing it at your client configuration and an existing directory for its sub-wallets:

walrus --config <PATH_TO_CLIENT_CONFIG> daemon \
    --bind-address "127.0.0.1:8899" \
    --sub-wallets-dir <SUB_WALLETS_DIR> \
    --n-clients 1

The script then uploads 1 MiB of random data with PUT /v1/blobs?epochs=5 and reads it back with GET /v1/blobs/<blob_id>. A first-time store returns a newlyCreated JSON object that carries the blob ID; storing data that already exists on Walrus returns an alreadyCertified object instead. See Storing Blobs for both response shapes.

examples/python/hello_walrus_webapi.py

# Copyright (c) Walrus Foundation
# SPDX-License-Identifier: Apache-2.0

# Example of uploading and downloading a file to / from the Walrus service
# using the walrus client web API facilities.
#
# Prerequisites:
#
# Run the Walrus client in daemon mode:
'''
../CONFIG/bin/walrus --config ../CONFIG/config_dir/client_config.yaml daemon \
    --bind-address 127.0.0.1:8899 \
    --sub-wallets-dir ../CONFIG/wallets \
    --n-clients 1
'''

# Std lib imports
import os
import time

# External requests HTTP library
import requests

ADDRESS = "127.0.0.1:8899"
EPOCHS = "5"


# Helper functions to upload a blob
def upload_blob(ADDRESS, EPOCHS, data):
    # Upload the data to the Walrus service  using a PUT request
    store_url = f"http://{ADDRESS}/v1/blobs?epochs={EPOCHS}"
    response = requests.put(store_url, data=data)

    # Assert the response status code
    assert response.status_code == 200
    blob_id = response.json()["newlyCreated"]["blobObject"]["blobId"]
    return blob_id


# Helper functions to download a blob
def download_blob(ADDRESS, blob_id):
    # Now read the same resource using the blob-id
    read_url = f"http://{ADDRESS}/v1/blobs/{blob_id}"
    response = requests.get(read_url)

    # Assert the response status code
    assert response.status_code == 200
    return response.content


# Upload a random 1MB string then download it, and check it matches
if __name__ == "__main__":
    # Generate a 1MB blob of random data
    random_data = os.urandom(1024 * 1024)

    # Upload the blob to the Walrus service
    start_time = time.time()
    blob_id = upload_blob(ADDRESS, EPOCHS, random_data)
    upload_time = time.time()

    # Now download the same blob using the blob-id
    data = download_blob(ADDRESS, blob_id)
    assert data == random_data
    download_time = time.time()

    # Print some information about the blob
    print(f"Blob ID: {blob_id}")
    print(f"Size {len(random_data)} bytes")
    print(f"Upload time: {upload_time - start_time:.2f}s")
    print(f"Download time: {download_time - upload_time:.2f}s")

The script prints the blob ID, the blob size, and the transfer times, similar to the following:

Blob ID: iIWkkUTzPZx-d1E_A7LqUynnYFD-ztk39_tP8MLdS2Y
Size 1048576 bytes
Upload time: 4.25s
Download time: 0.51s

To use public endpoints instead of a local daemon, send PUT requests to a Testnet publisher and GET requests to an aggregator from the Network Reference. Public endpoints limit requests to 10 MiB, and Walrus has no public unauthenticated publisher on Mainnet.

How to use the JSON API

The JSON API runs the walrus CLI as a subprocess and exchanges JSON on standard input and output. JSON mode covers every CLI command, which makes it the right path when you need operations the HTTP API does not expose, such as checking a blob's onchain certification. The example passes a JSON object that names the client configuration file and the command to run, for example:

{
  "config": "path/to/client_config.yaml",
  "command": {
    "store": {
      "files": ["some_file.bin"],
      "epochs": 2
    }
  }
}

The example stores a 1 MiB file, reads it back, and then checks the blob's certification. Every stored blob has a corresponding Sui object that records its storage lifetime and certification status, so the final part of the script fetches that object through Sui JSON-RPC (sui_getObject) and verifies that the onchain blob_id matches the uploaded blob.

examples/python/hello_walrus_jsonapi.py

# Copyright (c) Walrus Foundation
# SPDX-License-Identifier: Apache-2.0

# Example of uploading and downloading a file to / from the Walrus service
# Using the walrus client json input & output facilities.

# Std lib imports
import os
import subprocess
import json
import tempfile
import base64

import requests

from utils import num_to_blob_id, PATH_TO_WALRUS, PATH_TO_WALRUS_CONFIG, GRAPHQL_URL

try:
    # Create a 1MiB file of random data
    random_data = os.urandom(1024 * 1024)
    tmp = tempfile.NamedTemporaryFile(delete=False)
    tmp.write(random_data)
    tmp.close()

    # Part 1. Upload the file to the Walrus service
    store_json_command = f"""{{ "config" : "{PATH_TO_WALRUS_CONFIG}",
        "command" : {{ "store" :
        {{ "files" : ["{tmp.name}"], "epochs" : 2  }}}}
    }}"""
    result = subprocess.run(
        [PATH_TO_WALRUS, "json"],
        text=True,
        capture_output=True,
        input=store_json_command,
    )

    assert result.returncode == 0

    # Parse the response and display key information
    json_result_dict = json_result_dict = json.loads(result.stdout.strip())[0]["blobStoreResult"]
    if "newlyCreated" in json_result_dict:
        blob_id = json_result_dict["newlyCreated"]["blobObject"]["blobId"]
        sui_object_id = json_result_dict["newlyCreated"]["blobObject"]["id"]
    elif "alreadyCertified" in json_result_dict:
        blob_id = json_result_dict["alreadyCertified"]["blobId"]
    else:
        raise ValueError("Unexpected response from Walrus")

    print(
        f"Upload Blob ID: {blob_id} Size {len(random_data)} bytes"
    )

    if sui_object_id:
        print(f"Certificate in Object ID: {sui_object_id}")

    # Part 2. Download the file from the Walrus service
    read_json_command = f"""{{ "config" : "{PATH_TO_WALRUS_CONFIG}",
        "command" : {{ "read" :
        {{ "blobId" : "{blob_id}" }}}}
    }}"""
    result = subprocess.run(
        [PATH_TO_WALRUS, "json"],
        text=True,
        capture_output=True,
        input=read_json_command,
    )

    assert result.returncode == 0

    # Parse the response and display key information
    json_result_dict = json.loads(result.stdout.strip())
    downloaded_data = base64.b64decode(json_result_dict["blob"])
    assert downloaded_data == random_data

    print(
        f"Download Blob ID: {json_result_dict['blobId']} Size {len(downloaded_data)} bytes"
    )

    # Part 3. Check the availability of the blob
    if sui_object_id:
        query = """
        query ($objectId: SuiAddress!) {
            object(address: $objectId) {
                asMoveObject {
                    contents { json }
                }
            }
        }
        """
        response = requests.post(
            GRAPHQL_URL,
            json={"query": query, "variables": {"objectId": sui_object_id}},
        )
        object_content = response.json()["data"]["object"]["asMoveObject"][
            "contents"
        ]["json"]
        print("Object content:")
        print(json.dumps(object_content, indent=4))

        # Check that the blob ID matches the one we uploaded
        blob_id_downloaded = int(object_content["blob_id"])
        if num_to_blob_id(blob_id_downloaded) == blob_id:
            print("Blob ID matches certificate!")
        else:
            print("Blob ID does not match")

finally:
    os.unlink(tmp.name)

How to track Walrus events

Walrus coordinates storage through smart contracts on Sui, and blob lifecycle operations emit Sui events such as BlobRegistered and BlobCertified. Tracking these events lets you monitor activity, for example to confirm certification of your own blobs or to index recent stores. The script needs no walrus binary at runtime; it only reads the system object ID from your client configuration file and queries a Sui full node over JSON-RPC.

The script proceeds in three steps:

1. Reads the system_object ID from your Walrus client configuration.
2. Fetches that object with sui_getObject to discover the Walrus package ID.
3. Queries the latest 100 events emitted by the package's blob module with suix_queryEvents, then prints each event's timestamp, type, size (for registrations), blob ID, and transaction digest.

examples/python/track_walrus_events.py

# Copyright (c) Walrus Foundation
# SPDX-License-Identifier: Apache-2.0

# Track Walrus storage related events on the Sui blockchain

import datetime

# Std lib imports
import requests
import re

from utils import num_to_blob_id, GRAPHQL_URL, PATH_TO_WALRUS_CONFIG

system_object_id = re.findall(
    r"system_object:[ ]*(.*)", open(PATH_TO_WALRUS_CONFIG).read()
)[0]
print(f"System object ID: {system_object_id}")

# Query the Walrus system object on Sui through the GraphQL API
query = """
query ($objectId: SuiAddress!) {
    object(address: $objectId) {
        asMoveObject {
            contents { type { repr } }
        }
    }
}
"""
response = requests.post(
    GRAPHQL_URL,
    json={"query": query, "variables": {"objectId": system_object_id}},
)
assert response.status_code == 200

object_type = response.json()["data"]["object"]["asMoveObject"]["contents"]["type"][
    "repr"
]
walrus_package = re.findall("(0x[0-9a-f]+)::system", object_type)[0]
print(f"Walrus type: {walrus_package}")

# Query the latest 50 events emitted by the Walrus package (the maximum page size)
query = """
query ($eventType: String!) {
    events(last: 50, filter: { type: $eventType }) {
        nodes {
            timestamp
            contents { type { repr } json }
            transaction { digest }
        }
    }
}
"""
response = requests.post(
    GRAPHQL_URL,
    # Filtering by the `<package>::events` prefix matches all Walrus event types
    json={"query": query, "variables": {"eventType": f"{walrus_package}::events"}},
)
assert response.status_code == 200

events = response.json()["data"]["events"]["nodes"]
# Print the most recent events first
for event in reversed(events):
    # Parse the Walrus event
    tx_digest = event["transaction"]["digest"]
    event_type = event["contents"]["type"]["repr"].split("::")[-1]
    parsed_event = event["contents"]["json"]
    time_date = datetime.datetime.fromisoformat(
        event["timestamp"].replace("Z", "+00:00")
    )

    # Blob lifecycle events carry a blob ID; epoch events do not
    blob_id_num = parsed_event.get("blob_id")
    blob_id = num_to_blob_id(int(blob_id_num)) if blob_id_num is not None else ""

    # For registered blobs get their size in bytes
    if event_type == "BlobRegistered":
        size = f"{parsed_event['size']}B"
    else:
        size = ""

    print(f"{time_date} {event_type:<15} {size:>10} {blob_id} Tx:{tx_digest:<48}")