Docs
Get Started

Code examples

This page gives you a small client in Python and one in JavaScript. Each has one method for every endpoint in the API reference, and each handles authentication, URL-encoding of file names, paging, and errors, so your application code stays short.

Both read the credential from environment variables:

export KUBERNS_ACCESS_KEY_ID="KBS..."
export KUBERNS_SECRET="your-secret-access-key"

Keep these on your server, and never in code that runs in a browser. See Credentials.

Python

Install the one dependency:

pip install requests

Save the client as kuberns_buckets.py:

import os
from urllib.parse import quote
 
import requests
 
BASE_URL = "https://gateway.kuberns.cloud"
 
 
class BucketsError(Exception):
    def __init__(self, status, message):
        super().__init__(f"{status}: {message}")
        self.status = status
        self.message = message
 
 
def _message(response):
    try:
        body = response.json()
    except ValueError:
        return response.text or response.reason
 
    if isinstance(body.get("detail"), dict):
        return "; ".join(f"{field}: {', '.join(problems)}" for field, problems in body["detail"].items())
 
    return body.get("error") or body.get("message") or response.reason
 
 
class Buckets:
    def __init__(self, access_key_id=None, secret=None, base_url=BASE_URL):
        self.base_url = base_url.rstrip("/")
        self.session = requests.Session()
        self.session.auth = (
            access_key_id or os.environ["KUBERNS_ACCESS_KEY_ID"],
            secret or os.environ["KUBERNS_SECRET"],
        )
 
    def _request(self, method, path, **kwargs):
        response = self.session.request(method, self.base_url + path, timeout=60, **kwargs)
 
        if not response.ok:
            raise BucketsError(response.status_code, _message(response))
 
        return response
 
    def _data(self, method, path, **kwargs):
        return self._request(method, path, **kwargs).json()["data"]
 
    @staticmethod
    def _path(bucket, key=""):
        return f"/{bucket}/{quote(key, safe='/')}" if key else f"/{bucket}/"
 
    def list_buckets(self):
        return self._data("GET", "/buckets")
 
    def create_bucket(self, name):
        return self._data("POST", "/buckets", json={"name": name})
 
    def get_bucket(self, bucket):
        return self._data("GET", f"/buckets/{bucket}")
 
    def delete_bucket(self, bucket):
        self._data("DELETE", f"/buckets/{bucket}")
 
    def create_folder(self, bucket, folder_path):
        self._data("POST", f"/buckets/{bucket}/folders", json={"folder_path": folder_path})
 
    def get_visibility(self, bucket, key):
        return self._data("GET", f"/buckets/{bucket}/visibility", params={"key": key})
 
    def set_visibility(self, bucket, key, visibility):
        return self._data(
            "PUT", f"/buckets/{bucket}/visibility",
            json={"object_key": key, "visibility": visibility},
        )
 
    def get_link(self, bucket, key):
        return self._data("POST", f"/buckets/{bucket}/url", json={"object_key": key})
 
    def upload(self, bucket, key, body, content_type=None):
        headers = {"Content-Type": content_type} if content_type else {}
        return self._request("PUT", self._path(bucket, key), data=body, headers=headers).json()
 
    def download(self, bucket, key):
        return self._request("GET", self._path(bucket, key)).content
 
    def head(self, bucket, key):
        return self._request("HEAD", self._path(bucket, key)).headers
 
    def delete_object(self, bucket, key):
        self._request("DELETE", self._path(bucket, key))
 
    def list_page(self, bucket, prefix="", limit=None, token=None):
        params = {"prefix": prefix, "limit": limit, "token": token}
        params = {name: value for name, value in params.items() if value}
        return self._request("GET", self._path(bucket), params=params).json()
 
    def list_all(self, bucket, prefix=""):
        folders, objects, token = [], [], None
 
        while True:
            page = self.list_page(bucket, prefix=prefix, token=token)
            folders += page["folders"] or []
            objects += page["objects"] or []
            token = page["next_token"]
 
            if not token:
                return {"folders": folders, "objects": objects}

Create a bucket and keep its identifier

Do this once and keep the bucket value in your configuration. It is what you pass to every other call.

from kuberns_buckets import Buckets
 
api = Buckets()
 
bucket = api.create_bucket("product-assets")["bucket"]
print(bucket)  # kb-product-assets-3f9a1c2b7d4e

To find a bucket you created earlier:

for item in api.list_buckets()["buckets"]:
    print(item["name"], item["bucket"])

Upload and download

BUCKET = "kb-product-assets-3f9a1c2b7d4e"
 
# From bytes
api.upload(BUCKET, "notes/hello.txt", b"hello", "text/plain")
 
# From a file on disk
with open("summary.pdf", "rb") as f:
    api.upload(BUCKET, "reports/2026/summary.pdf", f, "application/pdf")
 
# Download
data = api.download(BUCKET, "reports/2026/summary.pdf")
with open("copy.pdf", "wb") as f:
    f.write(data)
 
# Check a file without downloading it
headers = api.head(BUCKET, "reports/2026/summary.pdf")
print(headers["Content-Length"], headers["Content-Type"])

List files and folders

list_all follows the paging for you and returns every folder and file directly inside a path:

top = api.list_all(BUCKET)
print(top["folders"])             # ['notes/', 'reports/']
 
reports = api.list_all(BUCKET, "reports/2026")
for item in reports["objects"]:
    print(item["key"], item["size"])

To fetch one page at a time, use list_page:

page = api.list_page(BUCKET, limit=50)
while True:
    for item in page["objects"] or []:
        print(item["key"])
    if not page["next_token"]:
        break
    page = api.list_page(BUCKET, limit=50, token=page["next_token"])

Folders

api.create_folder(BUCKET, "invoices")

Share a file

# Make a file public, and get its permanent link
shared = api.set_visibility(BUCKET, "logo.png", "public")
print(shared["url"])
 
# Make it private again
api.set_visibility(BUCKET, "logo.png", "private")
 
# Get a link of the right kind for a file, whatever its state
link = api.get_link(BUCKET, "invoices/2026-10.pdf")
print(link["url"], link["expires_at"])  # expires_at is None for a public file

A common use is letting a signed-in user download a private file without giving your credential to their browser. Your server asks for a temporary link and sends the user to it:

from flask import Flask, redirect
 
app = Flask(__name__)
api = Buckets()
 
@app.get("/invoices/<invoice_id>/download")
def download_invoice(invoice_id):
    link = api.get_link(BUCKET, f"invoices/{invoice_id}.pdf")
    return redirect(link["url"])

Delete

api.delete_object(BUCKET, "notes/hello.txt")
api.delete_bucket(BUCKET)  # deletes the bucket and everything in it

Handle errors

Every failed call raises BucketsError with the HTTP status and the message:

from kuberns_buckets import Buckets, BucketsError
 
api = Buckets()
 
try:
    api.download(BUCKET, "missing.txt")
except BucketsError as error:
    if error.status == 404:
        print("No such file")
    elif error.status == 402:
        print("Storage is unpaid. Top up your credits.")
    else:
        raise

JavaScript

This client needs Node.js 18 or later and has no dependencies, because it uses the built-in fetch. Save it as kuberns-buckets.mjs:

export const BASE_URL = "https://gateway.kuberns.cloud";
 
export class BucketsError extends Error {
  constructor(status, message) {
    super(`${status}: ${message}`);
    this.status = status;
    this.detail = message;
  }
}
 
async function messageFrom(response) {
  const text = await response.text();
 
  try {
    const body = JSON.parse(text);
 
    if (body.detail && typeof body.detail === "object") {
      return Object.entries(body.detail)
        .map(([field, problems]) => `${field}: ${problems.join(", ")}`)
        .join("; ");
    }
 
    return body.error || body.message || response.statusText;
  } catch {
    return text || response.statusText;
  }
}
 
const encodeKey = (key) => key.split("/").map(encodeURIComponent).join("/");
 
export class Buckets {
  constructor({
    accessKeyId = process.env.KUBERNS_ACCESS_KEY_ID,
    secret = process.env.KUBERNS_SECRET,
    baseUrl = BASE_URL,
  } = {}) {
    this.baseUrl = baseUrl.replace(/\/$/, "");
    this.authorization =
      "Basic " + Buffer.from(`${accessKeyId}:${secret}`).toString("base64");
  }
 
  async #request(method, path, { json, body, headers = {}, params } = {}) {
    const url = new URL(this.baseUrl + path);
 
    for (const [name, value] of Object.entries(params ?? {})) {
      if (value) url.searchParams.set(name, value);
    }
 
    const init = { method, headers: { Authorization: this.authorization, ...headers } };
 
    if (json !== undefined) {
      init.body = JSON.stringify(json);
      init.headers["Content-Type"] = "application/json";
    } else if (body !== undefined) {
      init.body = body;
    }
 
    const response = await fetch(url, init);
 
    if (!response.ok) {
      throw new BucketsError(response.status, await messageFrom(response));
    }
 
    return response;
  }
 
  async #data(method, path, options) {
    return (await (await this.#request(method, path, options)).json()).data;
  }
 
  #path(bucket, key = "") {
    return key ? `/${bucket}/${encodeKey(key)}` : `/${bucket}/`;
  }
 
  listBuckets() {
    return this.#data("GET", "/buckets");
  }
 
  createBucket(name) {
    return this.#data("POST", "/buckets", { json: { name } });
  }
 
  getBucket(bucket) {
    return this.#data("GET", `/buckets/${bucket}`);
  }
 
  async deleteBucket(bucket) {
    await this.#data("DELETE", `/buckets/${bucket}`);
  }
 
  async createFolder(bucket, folderPath) {
    await this.#data("POST", `/buckets/${bucket}/folders`, {
      json: { folder_path: folderPath },
    });
  }
 
  getVisibility(bucket, key) {
    return this.#data("GET", `/buckets/${bucket}/visibility`, { params: { key } });
  }
 
  setVisibility(bucket, key, visibility) {
    return this.#data("PUT", `/buckets/${bucket}/visibility`, {
      json: { object_key: key, visibility },
    });
  }
 
  getLink(bucket, key) {
    return this.#data("POST", `/buckets/${bucket}/url`, { json: { object_key: key } });
  }
 
  async upload(bucket, key, body, contentType) {
    const headers = contentType ? { "Content-Type": contentType } : {};
    const response = await this.#request("PUT", this.#path(bucket, key), { body, headers });
    return response.json();
  }
 
  async download(bucket, key) {
    const response = await this.#request("GET", this.#path(bucket, key));
    return Buffer.from(await response.arrayBuffer());
  }
 
  async head(bucket, key) {
    return (await this.#request("HEAD", this.#path(bucket, key))).headers;
  }
 
  async deleteObject(bucket, key) {
    await this.#request("DELETE", this.#path(bucket, key));
  }
 
  async listPage(bucket, { prefix, limit, token } = {}) {
    const response = await this.#request("GET", this.#path(bucket), {
      params: { prefix, limit, token },
    });
    return response.json();
  }
 
  async listAll(bucket, prefix = "") {
    const folders = [];
    const objects = [];
    let token;
 
    do {
      const page = await this.listPage(bucket, { prefix, token });
      folders.push(...(page.folders ?? []));
      objects.push(...(page.objects ?? []));
      token = page.next_token;
    } while (token);
 
    return { folders, objects };
  }
}

Create a bucket and keep its identifier

import { Buckets } from "./kuberns-buckets.mjs";
 
const api = new Buckets();
 
const { bucket } = await api.createBucket("product-assets");
console.log(bucket); // kb-product-assets-3f9a1c2b7d4e

To find a bucket you created earlier:

const { buckets } = await api.listBuckets();
for (const item of buckets) console.log(item.name, item.bucket);

Upload and download

import { readFile, writeFile } from "node:fs/promises";
 
const BUCKET = "kb-product-assets-3f9a1c2b7d4e";
 
// From a string
await api.upload(BUCKET, "notes/hello.txt", "hello", "text/plain");
 
// From a file on disk
await api.upload(BUCKET, "reports/2026/summary.pdf", await readFile("summary.pdf"), "application/pdf");
 
// Download
const data = await api.download(BUCKET, "reports/2026/summary.pdf");
await writeFile("copy.pdf", data);
 
// Check a file without downloading it
const headers = await api.head(BUCKET, "reports/2026/summary.pdf");
console.log(headers.get("content-length"), headers.get("content-type"));

List files and folders

listAll follows the paging for you:

const top = await api.listAll(BUCKET);
console.log(top.folders); // ['notes/', 'reports/']
 
const reports = await api.listAll(BUCKET, "reports/2026");
for (const item of reports.objects) console.log(item.key, item.size);

To fetch one page at a time, use listPage:

let token;
do {
  const page = await api.listPage(BUCKET, { limit: 50, token });
  for (const item of page.objects ?? []) console.log(item.key);
  token = page.next_token;
} while (token);

Folders

await api.createFolder(BUCKET, "invoices");

Share a file

// Make a file public, and get its permanent link
const shared = await api.setVisibility(BUCKET, "logo.png", "public");
console.log(shared.url);
 
// Make it private again
await api.setVisibility(BUCKET, "logo.png", "private");
 
// Get a link of the right kind for a file, whatever its state
const link = await api.getLink(BUCKET, "invoices/2026-10.pdf");
console.log(link.url, link.expires_at); // expires_at is null for a public file

A common use is letting a signed-in user download a private file without giving your credential to their browser. Your server asks for a temporary link and sends the user to it:

import express from "express";
 
const app = express();
const api = new Buckets();
 
app.get("/invoices/:id/download", async (req, res) => {
  const link = await api.getLink(BUCKET, `invoices/${req.params.id}.pdf`);
  res.redirect(link.url);
});

Delete

await api.deleteObject(BUCKET, "notes/hello.txt");
await api.deleteBucket(BUCKET); // deletes the bucket and everything in it

Handle errors

Every failed call throws a BucketsError with the HTTP status in status and the message in detail:

import { Buckets, BucketsError } from "./kuberns-buckets.mjs";
 
try {
  await api.download(BUCKET, "missing.txt");
} catch (error) {
  if (!(error instanceof BucketsError)) throw error;
 
  if (error.status === 404) console.log("No such file");
  else if (error.status === 402) console.log("Storage is unpaid. Top up your credits.");
  else throw error;
}

Other languages

Any language with an HTTP client can do the same. Every call is one request with Basic authentication. See the API reference for the request and response of each endpoint.