Python SDK

pip install blobrouter

from blobrouter import BlobRouter

storage = BlobRouter(
    api_key="br_live_xxx",
    base_url="https://api.blobrouter.com",  # optional
    max_retries=3,  # optional, default 3
)

import os
from blobrouter import BlobRouter

storage = BlobRouter(api_key=os.environ["BLOBROUTER_API_KEY"])

with open("model_checkpoint.pt", "rb") as f:
    result = storage.upload(
        file=f,
        file_name="model_checkpoint.pt",
        content_type="application/octet-stream",
        file_size_bytes=os.path.getsize("model_checkpoint.pt"),
        priority="archive",
    )

print(result["file_id"], result["provider"], result["saved_vs_aws"])

import io
import os
import torch
from blobrouter import BlobRouter

storage = BlobRouter(api_key=os.environ["BLOBROUTER_API_KEY"])

buffer = io.BytesIO()
torch.save(model.state_dict(), buffer)
buffer.seek(0)

result = storage.upload(
    file=buffer,
    file_name="checkpoint_epoch_10.pt",
    content_type="application/octet-stream",
    file_size_bytes=buffer.getbuffer().nbytes,
    priority="archive",
)
print(result["file_id"], result["saved_vs_aws"])

got = storage.get(result["file_id"])
print(got["url"], got["expires_at"])

storage.delete(result["file_id"])

from blobrouter import BlobRouter, BlobRouterError

try:
    storage.upload(...)
except BlobRouterError as err:
    print(err, err.code, err.status_code)