Blob Operations Quickstart
This quickstart shows the core blob operations, store, read, and check status, three ways: the Walrus CLI, the HTTP API, and Python. Each operation appears in all three so you can pick the surface that fits your stack, or compare them side by side.
Walrus does not ship a dedicated Python SDK. A Python backend integrates with Walrus through the HTTP API, reading from an aggregator and writing to a publisher, or by driving the walrus CLI as a subprocess. This page uses the HTTP API for the Python examples because it needs nothing beyond the requests library.
- Prerequisites
- No API key required. The public Walrus aggregator and publisher endpoints are open.
- Set the endpoints you want to use. The examples below use the Testnet endpoints from the Network Reference:
export AGGREGATOR=https://aggregator.walrus-testnet.walrus.space
export PUBLISHER=https://publisher.walrus-testnet.walrus.space
- Choose a surface and complete the necessary setup:
- CLI: Install and configure the
walrusclient. See Getting Started. - HTTP API: Any HTTP client, such as
curl. - Python: Install the requests library with
pip install requests.
- CLI: Install and configure the
Public aggregators and publishers limit requests to 10 MiB, and Walrus does not provide a public unauthenticated publisher on Mainnet. For larger blobs or production Mainnet writes, run your own publisher, use an upload relay, or drive the CLI. See the Mainnet Publisher Production Guide.
Store a blob
Store a file for a number of epochs. The store returns a blob ID.
- CLI
- HTTP API
- Python
$ walrus store ./report.pdf --epochs 5
$ curl -X PUT "$PUBLISHER/v1/blobs?epochs=5" --upload-file ./report.pdf
The HTTP API returns JSON. A first-time store returns a newlyCreated object, and a store of a blob that already exists returns an alreadyCertified object. The blob ID is in both.
import requests
PUBLISHER = "https://publisher.walrus-testnet.walrus.space"
def store_blob(data: bytes, epochs: int = 5) -> str:
response = requests.put(
f"{PUBLISHER}/v1/blobs",
params={"epochs": epochs},
data=data,
)
response.raise_for_status()
result = response.json()
# A first-time store returns "newlyCreated"; an existing blob returns
# "alreadyCertified". The blob ID is available in both cases.
if "newlyCreated" in result:
return result["newlyCreated"]["blobObject"]["blobId"]
return result["alreadyCertified"]["blobId"]
with open("report.pdf", "rb") as file:
blob_id = store_blob(file.read())
print(f"stored blob {blob_id}")
Read a blob
Read a blob using its blob ID.
- CLI
- HTTP API
- Python
$ walrus read <BLOB_ID> --out ./report.pdf
Without the flag --out, the CLI writes the blob to standard output.
$ curl "$AGGREGATOR/v1/blobs/<BLOB_ID>" -o ./report.pdf
import requests
AGGREGATOR = "https://aggregator.walrus-testnet.walrus.space"
def read_blob(blob_id: str) -> bytes:
response = requests.get(f"{AGGREGATOR}/v1/blobs/{blob_id}")
response.raise_for_status()
return response.content
data = read_blob(blob_id)
with open("report.pdf", "wb") as file:
file.write(data)
print(f"read {len(data)} bytes")
When you read a blob immediately after certifying it, a public aggregator can briefly return 404 before its read path catches up. Retry with backoff in that window, as the full Python example shows. See Reading blobs after upload for more.
Check a blob's status
Check whether a blob is available and its registered onchain information.
- CLI
- HTTP API
- Python
$ walrus blob-status --blob-id <BLOB_ID>
The CLI reports the blob's certification status and end epoch from its Sui object. See Reading blobs to learn more about certification status and Sui objects.
The aggregator serves reads, so a successful GET confirms availability. A 200 means you can read the blob. A 404 immediately after you certified the blob is usually transient on public, CDN-fronted aggregators, so retry with backoff; a 404 for a blob you did not just write is final. Treat 503 and other 5xx responses as retryable.
$ curl -o /dev/null -s -w "%{http_code}\n" "$AGGREGATOR/v1/blobs/<BLOB_ID>"
For the onchain certification status and end epoch, use the CLI blob-status command.
import requests
AGGREGATOR = "https://aggregator.walrus-testnet.walrus.space"
def read_status(blob_id: str) -> str:
# A 404 right after certification is usually transient on public
# aggregators; a 404 for a blob you did not just write is final. Treat any
# 5xx as transient too.
response = requests.get(f"{AGGREGATOR}/v1/blobs/{blob_id}")
if response.status_code == 200:
return "available"
if response.status_code == 404 or 500 <= response.status_code < 600:
return "retry"
return "unavailable"
print(read_status(blob_id)) # "available", "retry", or "unavailable"
Full Python example
The following script stores a blob, reads it back, and verifies the round trip against a public Testnet publisher and aggregator.
import os
import time
import requests
AGGREGATOR = "https://aggregator.walrus-testnet.walrus.space"
PUBLISHER = "https://publisher.walrus-testnet.walrus.space"
def store_blob(data: bytes, epochs: int = 5) -> str:
response = requests.put(
f"{PUBLISHER}/v1/blobs", params={"epochs": epochs}, data=data
)
response.raise_for_status()
result = response.json()
if "newlyCreated" in result:
return result["newlyCreated"]["blobObject"]["blobId"]
return result["alreadyCertified"]["blobId"]
def read_blob(blob_id: str, attempts: int = 5, backoff: float = 1.0) -> bytes:
# A public aggregator can briefly return 404 or a 5xx right after a blob is
# certified, before its read path catches up. Because you know the blob was
# just stored, retry those statuses with backoff before giving up.
response = None
for attempt in range(attempts):
response = requests.get(f"{AGGREGATOR}/v1/blobs/{blob_id}")
if response.status_code == 200:
return response.content
retryable = response.status_code == 404 or 500 <= response.status_code < 600
if not retryable:
break
time.sleep(backoff * (2**attempt))
response.raise_for_status()
return response.content
if __name__ == "__main__":
original = os.urandom(1024)
blob_id = store_blob(original)
print(f"stored blob {blob_id}")
downloaded = read_blob(blob_id)
assert downloaded == original, "round trip mismatch"
print(f"read {len(downloaded)} bytes back, round trip verified")
For more Python examples, including driving the CLI through its JSON interface and tracking Walrus events, see Using Walrus with Python.
References
- Store blobs with the CLI: Store blobs with the Walrus client
- Use the HTTP API in depth: Storing blobs with the HTTP API and Reading blobs with the HTTP API
- Run a publisher that pays for storage on behalf of your users: Mainnet Publisher Production Guide
- Canonical endpoints and IDs: Network Reference