Python SDK

The apulodi package mirrors @apulodi/sdk endpoint-for-endpoint: the same resources, the same auth, the same error envelope — in idiomatic Python. It ships as a single dependency (httpx) and is fully typed.

Server-side only. Your API key is a secret credential. Use the SDK from servers, scripts and jobs — never from browser code.

Install

bash
pip install apulodi

Requires Python 3.9+. The only runtime dependency is httpx.

Configure

python
import os
from apulodi import Apulodi

apulodi = Apulodi(api_key=os.environ["APULODI_API_KEY"])

For local development against your own instance:

python
apulodi = Apulodi(
    api_key=os.environ["APULODI_API_KEY"],
    base_url="http://localhost:3000",
)
OptionTypeDefaultDescription
api_keystr— (required)Project-scoped API key (apk_test_… / apk_live_…)
base_urlstrhttps://app.apulodi.devAPI origin
timeoutfloat30.0Per-request timeout in seconds (storage PUTs get 4×)

The client rejects an empty api_key and non-http(s) base_url at construction, so misconfiguration fails fast. Verify the key with apulodi.me() — it calls GET /v1/me and returns the project / organization / API-key context.

Upload a file

python
from apulodi import Apulodi

apulodi = Apulodi(api_key=os.environ["APULODI_API_KEY"])

# One call performs the full flow: initiate → direct-to-storage PUT → complete.
file = apulodi.files.upload(
    open("avatar.jpg", "rb"),          # bytes / bytearray / binary file object
    file_name="avatar.jpg",
    content_type="image/jpeg",
    path="users/avatars",              # logical folder, created automatically
    metadata={"userId": "u_123"},
)
print(file["id"], file["status"])      # file_… uploaded

This method performs three API calls:

  1. POST /v1/files/upload → PENDING file + presigned URL
  2. PUT the bytes directly to storage — never through APULODI's servers
  3. POST /v1/files/:id/complete → server verifies size and content type

Accepted inputs: bytes, bytearray, memoryview, or any binary file object (read fully into memory — the byte size is part of the signed request). Files over the multipart threshold (8 MiB by default, tune with multipart_threshold=) automatically use the multipart flow — and the session is aborted cleanly if a part fails.

Retry-safe uploads pass an idempotency key:

python
apulodi.files.upload(data, "report.pdf", idempotency_key="order-1234")

Repeating the same key returns the same file instead of creating a duplicate.

Files

python
# Metadata
file = apulodi.files.get("file_…")
file["filename"]; file["contentType"]; file["status"]; file["metadata"]

# Listing with filters (they mirror the REST query params) + cursor pagination
page = apulodi.files.list({"path": "users/avatars", "limit": 50, "search": "avatar"})
page["pagination"]["nextCursor"]      # pass as {"cursor": …} for the next page

# …or iterate over everything matching a filter — cursors handled for you
for f in apulodi.files.iterate({"contentType": "image/png"}):
    print(f["filename"], f["size"])

# Rename / move / metadata (logical-only — no bytes are copied)
apulodi.files.update("file_…", {"filename": "profile.jpg", "path": "users/profiles"})

# Server-side copy (returns a new file id)
apulodi.files.copy("file_…", path="backups", filename="profile-backup.jpg")

# Short-lived presigned download URL — bytes go straight from storage
signed = apulodi.files.download_url("file_…", expires_in_seconds=300)
signed["url"]; signed["expiresAt"]

Replace content (versioning)

python
v2 = apulodi.files.replace("file_…", new_bytes, content_type="image/jpeg")
v2["version"]  # 2

replace() runs the initiate → PUT → complete steps for you; the previous version stays intact server-side.

Delete & restore

python
apulodi.files.delete("file_…")    # {"id": …, "deleted": True} — soft-delete
back = apulodi.files.restore("file_…")  # re-activates within the grace window

Deletion soft-deletes the file: it disappears from listings and reads return 404, but the storage object is kept for a grace window (default 7 days) before a purge sweep removes it. restore() re-activates the file while the object still exists; afterwards it fails with a typed ApulodiError (409, OBJECT_NOT_FOUND).

Image & media transforms

python
# Fire-and-forget: returns immediately with a pending variant. Idempotent.
variant = apulodi.files.transform(
    "file_…", width=256, height=256, fit="cover", format="webp", quality=80
)

# Poll until ready (or subscribe to the file.processed webhook event).
done = apulodi.files.wait_for_variant("file_…", variant["id"], timeout=30)

# List every variant of a file / download the derivative from storage.
all_variants = apulodi.files.variants("file_…")
signed = apulodi.files.variant_download_url("file_…", done["id"], expires_in_seconds=3600)

The same API handles video and audio: transcode with format="mp4" / "webm", extract a poster frame with poster=True, convert audio with format="mp3" | "m4a" | "ogg" | "wav". Full reference: Media transformation.

Multipart (explicit control)

python
result = apulodi.uploads.create_multipart("big.mp4", "video/mp4", size=250_000_000)
session, pending = result["session"], result["file"]

parts = []
with open("big.mp4", "rb") as f:
    for part_meta in session["parts"]:
        chunk = f.read(session["partSize"])
        parts.append(apulodi.uploads.upload_part(session, part_meta["partNumber"], chunk))

file = apulodi.uploads.complete(session["id"], parts)

upload_part() uses the session's presigned URL and mints a fresh one automatically if the URL is missing or expired. abort(session_id) discards uploaded parts (idempotent), and list_stored_parts(session_id) shows what storage currently holds.

Folders & usage

python
apulodi.folders.list("users")             # subfolders of a logical path
usage = apulodi.usage.get("2026-09")      # omit the period for this month
usage["storage"]["usedBytes"]

Webhooks

Register endpoints that receive signed POST deliveries of APULODI events:

python
result = apulodi.webhooks.create("https://example.com/hooks/apulodi")
secret = result["secret"]         # shown EXACTLY ONCE — store it server-side

apulodi.webhooks.list()
apulodi.webhooks.deliveries("wh_…", limit=25)   # status / attempts / lastError
apulodi.webhooks.redeliver("wh_…", "dl_…")

Verify deliveries in your receiving endpoint before trusting them. The verifier parses t=<unix>,v1=<hex> headers, compares in constant time and enforces a 5-minute replay window — byte-compatible with deliveries signed by the APULODI platform:

python
import json
import os

from apulodi import verify_webhook_signature

# Flask example — verify over the RAW body, before any parsing:
@app.post("/webhooks/apulodi")
def apulodi_webhook():
    raw = request.get_data()
    signature = request.headers.get("APULODI-Signature", "")

    if not verify_webhook_signature(
        os.environ["APULODI_WEBHOOK_SECRET"], raw.decode("utf-8"), signature
    ):
        return {"error": "invalid signature"}, 401

    event = json.loads(raw)
    # event["type"] == "file.uploaded", event["data"]["file"]["id"], …
    return {"ok": True}

Respond 2xx quickly (APULODI times out at 10s and retries on failure).

Errors

Every non-2xx response, network failure and timeout raises ApulodiError:

python
from apulodi import ApulodiError

try:
    apulodi.files.get("file_missing")
except ApulodiError as e:
    e.status            # 404
    e.code              # "FILE_NOT_FOUND"
    e.message           # human-readable
    e.details           # structured extras, if the API sent any
    e.is_client_error   # True — the request was wrong, retrying won't help
    e.is_server_error   # True — retrying later may help

Network failures map to code: "NETWORK_ERROR" and timeouts to "TIMEOUT" (both with status: 0). API keys are always scrubbed from error messages, and raw provider/database errors are never included. See Error handling for the full behavior.

REST mapping

SDKREST API
apulodi.me()GET /v1/me
apulodi.files.list(params) / iterate(params)GET /v1/files
apulodi.files.get(file_id)GET /v1/files/:id
apulodi.files.upload(file, file_name=…)POST /v1/files/upload + direct PUT + POST /v1/files/:id/complete
apulodi.files.complete(file_id)POST /v1/files/:id/complete
apulodi.files.update(file_id, params)PATCH /v1/files/:id
apulodi.files.delete(file_id)DELETE /v1/files/:id
apulodi.files.copy(file_id, path=…, filename=…)POST /v1/files/:id/copy
apulodi.files.download_url(file_id, expires_in_seconds=…)POST /v1/files/:id/download-url
apulodi.files.restore(file_id)POST /v1/files/:id/restore
apulodi.files.replace(file_id, data, content_type)POST /v1/files/:id/replace + PUT + POST /v1/files/:id/replace/complete
apulodi.files.transform(file_id, …)POST /v1/files/:id/transform
apulodi.files.variants(file_id)GET /v1/files/:id/variants
apulodi.files.wait_for_variant(file_id, variant_id)GET /v1/files/:id/variants (polling)
apulodi.files.variant_download_url(file_id, variant_id)POST /v1/files/:id/variants/:variantId/download-url
apulodi.uploads.create_multipart(…)POST /v1/uploads/multipart
apulodi.uploads.upload_part(…) / regenerate_part_urls(…)POST /v1/uploads/multipart/:id/parts
apulodi.uploads.list_stored_parts(…)GET /v1/uploads/multipart/:id/parts
apulodi.uploads.complete(…)POST /v1/uploads/multipart/:id/complete
apulodi.uploads.abort(…)DELETE /v1/uploads/multipart/:id
apulodi.folders.list(path)GET /v1/folders
apulodi.usage.get(period)GET /v1/usage
apulodi.webhooks.create(url, events=…)POST /v1/webhooks
apulodi.webhooks.list() / get() / delete()GET / GET / DELETE /v1/webhooks(:id)
apulodi.webhooks.deliveries(…)GET /v1/webhooks/:id/deliveries
apulodi.webhooks.redeliver(…)POST /v1/webhooks/:id/deliveries/:deliveryId/redeliver

Next: SDK — Files.