Python SDK

steadylink is the official Python client for the SteadyLink API. It is synchronous, uses only the standard library, and is a good fit for Django or Flask views, background workers, data pipelines, and one-off scripts. This page lists every public method with its exact arguments and what it sends, so you can predict what each call does before you run it.

Version 0.1.0 covers uploads, replacements, and the platform routes. For anything it does not wrap, such as listing files or minting signed links, use request() with any route from the REST API.

Install#

Requires Python 3.10 or later.

python -m pip install steadylink

The package includes a py.typed marker, so mypy and Pyright pick up its inline annotations.

Create a client#

import os
from steadylink import SteadyLink

client = SteadyLink(api_key=os.environ["STEADYLINK_API_KEY"])

buckets = client.list_buckets()["items"]

Create a key in Dashboard > Developers with assets:read and assets:write. Keep it on the server: in environment variables or a secret manager, never in source control or a client app.

All arguments are keyword-only. The constructor raises ValueError when you pass both api_key and access_token, or neither.

The default transports use urllib. JSON calls time out after 30 seconds and presigned uploads after 60 seconds. A large file on a slow connection can exceed 60 seconds; supply your own binary_transport with a longer timeout for big uploads.

Every request sends Accept: application/json and User-Agent: steadylink-python/0.1.0. Methods return the decoded JSON response as plain dict and list values.

Uploads#

upload_file#

Creates an upload session, sends the bytes to the presigned storage URL, and completes the session.

upload_file(bucket_id: str, filename: str, body: bytes, content_type: str = "application/octet-stream", path: str = "") -> dict

Returns the completed upload session. Completion queues finalization, so the session's status is usually committing and objectAssetId may still be null. Poll GET /api/upload-batches/{batchId} with request() until the session is ready to get the asset ID. Uploading to a key that already exists adds a new revision to that file.

from pathlib import Path

path = Path("campaign-hero.webp")
session = client.upload_file(
    "8b1f6c2e-4a7d-4f0e-b3c9-2d5e7a9f1c04",
    filename=path.name,
    body=path.read_bytes(),
    path="campaign/",
)
print(session["id"], session["status"])

create_upload_batch#

Creates up to 100 upload sessions in one call. Each session in the response's files list has an id and a presigned uploadUrl.

create_upload_batch(bucket_id: str, files: list[UploadFile]) -> dict

UploadFile is a frozen dataclass exported by the package:

@dataclass(frozen=True)
class UploadFile:
    filename: str
    size: int
    contentType: str = "application/octet-stream"
    path: str = ""

Route: POST /api/upload-batches.

upload_to_url#

Uploads bytes to a presigned URL with a PUT. It never attaches SteadyLink credentials, so it is safe to use with URLs your server hands out.

upload_to_url(upload_url: str, body: bytes, content_type: str = "application/octet-stream") -> None

Raises SteadyLinkError for a non-2xx response from storage and SteadyLinkNetworkError when storage cannot be reached. It is not retried.

complete_upload#

complete_upload(session_id: str, idempotency_key: str | None = None) -> dict

Route: POST /api/upload-sessions/{session_id}/complete. Sends Idempotency-Key: complete-{session_id} unless you pass a key, which makes it safe to retry and lets the client retry it automatically on 429, 502, 503, and 504.

cancel_upload#

cancel_upload(session_id: str) -> None

Route: DELETE /api/upload-sessions/{session_id}. Deletes the temporary bytes and releases the reserved storage. Sessions already ready, blocked, or failed return 409.

Upload with a specific content type#

To store a type other than application/octet-stream, declare it on the session and send the bytes with the default type:

from pathlib import Path
from steadylink import UploadFile

path = Path("campaign-hero.webp")
data = path.read_bytes()

batch = client.create_upload_batch(bucket_id, [UploadFile(path.name, len(data), "image/webp", "campaign/")])
session = batch["files"][0]
client.upload_to_url(session["uploadUrl"], data)   # Content-Type: application/octet-stream
client.complete_upload(session["id"])

Replacements#

A replacement publishes new bytes as the next revision of an existing file. The asset ID and every link to it stay the same.

replace_file#

replace_file(bucket_id: str, key: str, filename: str, body: bytes, content_type: str = "application/octet-stream") -> dict

Calls create_replacement_upload(), sends the bytes with upload_to_url(), then calls finish_replacement(). Returns {"replaced": True, "version": 4}. Leave content_type at its default (see the warning under Uploads). To declare a specific type for the new revision, run the three steps yourself:

pending = client.create_replacement_upload(bucket_id, len(data), "image/webp")
client.upload_to_url(pending["uploadUrl"], data)    # Content-Type: application/octet-stream
client.finish_replacement(bucket_id, "campaign/campaign-hero.webp", pending["tempKey"], "campaign-hero-v2.webp")

from pathlib import Path

replacement = Path("campaign-hero-v2.webp")
result = client.replace_file(
    bucket_id,
    "campaign/campaign-hero.webp",
    filename=replacement.name,
    body=replacement.read_bytes(),
)
print(result["version"])

A file caught by the malware scan before commit is rejected with 400 and the current revision keeps serving. If the background scan finds malware after the new revision went current, the link returns 404 until you restore a clean revision. See The scan window after a replacement.

create_replacement_upload#

create_replacement_upload(bucket_id: str, size: int, content_type: str = "application/octet-stream") -> dict

Route: POST /api/assets/{bucket_id}/objects/upload-temp. Returns {"uploadUrl": ..., "tempKey": ...}.

finish_replacement#

finish_replacement(bucket_id: str, key: str, temp_key: str, original_filename: str | None = None) -> dict

Route: POST /api/assets/{bucket_id}/objects/replace. Commits the temporary bytes as the next revision of key.

Buckets and assets#

list_buckets#

list_buckets(limit: int = 100) -> dict

Route: GET /api/assets/. Returns {"items": [...], "nextCursor": ...}. The method has no cursor argument; to page past the first limit buckets, call request("GET", f"/api/assets/?limit=100&cursor={cursor}").

get_asset#

get_asset(asset_id: str) -> dict

Route: GET /api/assets/{asset_id}. Reads a bucket or asset record.

Platform helpers#

These wrap routes under /api/platform. With an API key, reads need workspace:read and writes need workspace:admin, including create_migration() and set_focal_point(). Field details are in the Platform API reference.

create_migration#

create_migration(bucket_id: str, files: list[UploadFile]) -> dict

Creates a migration record with an upload batch of up to 100 sessions and returns {"id", "status", "uploadBatch"}. Send and complete each session in uploadBatch["files"] as above. A migration.completed webhook fires when the batch finishes.

list_migrations#

list_migrations() -> dict

The 50 most recent migrations, each with id, bucketId, uploadBatchId, status, totalFiles, and createdAt.

set_focal_point#

set_focal_point(asset_id: str, x: float, y: float) -> dict

Sets the default crop position for cover transforms. x and y run from 0 to 1. Returns {"x": 0.42, "y": 0.31}.

list_delivery_domains#

list_delivery_domains() -> dict

add_delivery_domain#

add_delivery_domain(hostname: str) -> dict

Returns the domain with the TXT record to create at _steadylink.{hostname}.

verify_delivery_domain#

verify_delivery_domain(domain_id: str) -> dict

get_delivery_policy#

get_delivery_policy() -> DeliveryPolicy

Returns {"mode", "countries", "storageRegion", "availableStorageRegions"}.

set_delivery_policy#

set_delivery_policy(policy: DeliveryPolicy) -> DeliveryPolicy

DeliveryPolicy is a TypedDict exported by the package: mode ("all", "allow", or "deny") and countries (two-letter ISO codes) are required, storageRegion is optional and can only be the current region.

client.set_delivery_policy({"mode": "deny", "countries": ["KP", "RU"]})

create_webhook#

create_webhook(url: str, events: list[str]) -> dict

Registers a generic HTTPS receiver for up to 10 event types. The response includes the signing secret, returned only once. For a Discord, Slack, or Teams destination, call request() with destinationType; see Webhooks.

list_webhooks#

list_webhooks() -> dict

Returns {"availableEvents": [...], "items": [...]}.

test_webhook#

test_webhook(webhook_id: str) -> dict

Sends a test event and waits for your endpoint. Returns {"delivered": True, "responseStatus": 200, "deliveryId": ...}; a non-2xx answer raises SteadyLinkError with status 502, and api_error_code(error) returns webhook_delivery_failed.

Low-level calls#

request#

Sends an authenticated JSON request to any route. Every other method is built on it.

request(method: str, path: str, payload: dict | None = None, *, idempotency_key: str | None = None, retry: bool = True) -> Any

payload is sent as compact JSON with Content-Type: application/json. idempotency_key adds an Idempotency-Key header and makes a write eligible for retries. retry=False turns retries off for this call. Returns the decoded JSON, the text for a non-JSON body, or None for an empty body.

signed = client.request("POST", f"/api/assets/{asset_id}/signed-url?ttl=3600&name=client+preview")
print(signed["token"], signed["expiresAt"])

Errors#

Both error classes subclass RuntimeError and are exported from steadylink.

SteadyLinkError#

A non-success HTTP response from the API or from storage.

str(error) is the body's message when it is a string, such as Upload session expired, else SteadyLink request failed with HTTP {status}.

def api_error_code(error: SteadyLinkError) -> str | None:
    message = error.detail.get("message") if isinstance(error.detail, dict) else None
    return message.get("code") if isinstance(message, dict) else None

SteadyLinkNetworkError#

The request never got an HTTP response: DNS failure, refused connection, or timeout. The original exception is chained as __cause__.

from steadylink import SteadyLinkError, SteadyLinkNetworkError

try:
    client.list_buckets()
except SteadyLinkError as error:
    print(error.status, api_error_code(error), error.request_id)
except SteadyLinkNetworkError as error:
    print(str(error), error.__cause__)

Status codes and error codes are listed on Errors.

Retries#

A request is retried up to max_retries times when both of these hold:

• The method is GET, HEAD, or OPTIONS, or the call passed an idempotency_key. complete_upload() always does.
• The request raised a network error, or the response was 429, 502, 503, or 504.

The client waits for the Retry-After value when the response has one, otherwise 0.25 s, 0.5 s, 1 s, and so on, each with up to 0.1 s of random jitter. Other writes and presigned uploads are never retried automatically, because repeating them could apply a change twice.

Next steps#