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
pip install apulodi
Requires Python 3.9+. The only runtime dependency is httpx.
Configure
import os
from apulodi import Apulodi
apulodi = Apulodi(api_key=os.environ["APULODI_API_KEY"])
For local development against your own instance:
apulodi = Apulodi(
api_key=os.environ["APULODI_API_KEY"],
base_url="http://localhost:3000",
)
| Option | Type | Default | Description |
|---|---|---|---|
api_key | str | — (required) | Project-scoped API key (apk_test_… / apk_live_…) |
base_url | str | https://app.apulodi.dev | API origin |
timeout | float | 30.0 | Per-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
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:
POST /v1/files/upload→ PENDING file + presigned URLPUTthe bytes directly to storage — never through APULODI's serversPOST /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:
apulodi.files.upload(data, "report.pdf", idempotency_key="order-1234")
Repeating the same key returns the same file instead of creating a duplicate.
Files
# 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)
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
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
# 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)
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
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:
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:
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:
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
| SDK | REST 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.