xlambda-media

Image, video, and 360°-tour upload and transforms, plus live streaming (RTMP/SRT/WHIP ingest, WHEP/HLS playback).

pip install xlambda-media

Python port of @xlambda-tech/media.

Uploading

The platform's upload is a three-call dance — presign, PUT the bytes to storage, confirm. This collapses it into one call:

from xlambda.media import MediaClient

client = MediaClient(api_key=os.environ["XLAMBDA_API_KEY"])

result = client.media.upload(
    file=Path("photo.jpg"),          # bytes, a path, or any binary file object
    filename="photo.jpg",
    content_type="image/jpeg",
    project_id="proj_123",
)
print(result["mediaId"])

A path is streamed rather than read into memory, which matters for video. The size is computed for you.

If the transfer to storage fails, you get an UploadTransferError carrying the media_id:

from xlambda.media import UploadTransferError

try:
    client.media.upload(file=data, filename="a.mp4", content_type="video/mp4")
except UploadTransferError as err:
    log.warning("orphaned pending media %s: %s", err.media_id, err)

That id matters — the presign already created a row server-side, and there's no automatic cleanup, so losing the id loses the record.

Batches presign in one round trip and report per-file outcomes, index-aligned with the input, so one bad file doesn't sink the rest:

results = client.media.upload_batch(files=[
    {"file": a, "filename": "a.jpg", "contentType": "image/jpeg"},
    {"file": b, "filename": "b.jpg", "contentType": "image/jpeg"},
])
# [{"success": True, "mediaId": ...}, {"success": False, "error": ...}]

The sync version transfers sequentially; AsyncMediaClient.media.upload_batch runs them concurrently via asyncio.gather, matching the npm SDK.

upload_stream() is the one-step alternative — bytes go through the platform as multipart instead of direct-to-storage, with no confirm step. Prefer upload() for anything client-originated; use this for server-to-server piping where you already hold the bytes.

Image transforms

A pure URL builder — no network call, no auth header, because there's nowhere to put one on an <img> tag:

from xlambda.media import transform_url

client.image.transform_url("med_123", width=800, format="webp", quality=90)
# https://api.xlambda.tech/v1/image/med_123?width=800&format=webp&quality=90

crop="fill" requires both width and height, and raises ValueError if you omit one rather than sending a request that would fail server-side.

Live streaming

stream = client.livestreams.create(project_id="proj_123", title="Launch", record_enabled=True)

stream["rtmpServerUrl"], stream["rtmpStreamKey"]   # paste into OBS
stream["hlsPlaybackUrl"]                           # for players
stream["webRtcIngestUrl"], stream["webRtcPlaybackUrl"]   # WHIP / WHEP

The ingest secrets come back only from create() and regenerate_key() — the plaintext stream key is never retrievable again. regenerate_key() kicks the current publisher immediately and needs the admin scope.

Simulcast and clips hang off a stream id:

client.livestreams.simulcast_targets(stream["id"]).create(
    label="YouTube", destination_url="rtmp://a.rtmp.youtube.com/live2/KEY"
)
client.livestreams.clips(stream["id"]).create(duration_seconds=30)

WHIP/WHEP are URLs for a browser or player to use directly; this SDK doesn't wrap the SDP exchange.

Webhooks

from xlambda.media import create_media_events_handler
from xlambda.core.webhooks.adapters.flask import create_flask_blueprint

handler = create_media_events_handler(
    secret=os.environ["XLAMBDA_WEBHOOK_SECRET"],
    on_ready=lambda data: Media.objects.filter(pk=data["mediaId"]).update(status="ready"),
    on_failed=lambda data: log.error("processing failed: %s", data["error"]),
    on_livestream_recorded=lambda data: attach_recording(data["liveStreamId"], data["mediaId"]),
)

app.register_blueprint(create_flask_blueprint(handler), url_prefix="/webhooks/media")

Every callback is optional, and anything that isn't a media.*/livestream.* event is acknowledged with a 200 and ignored — so pointing a shared webhook URL here is safe. create_async_media_events_handler is the FastAPI-side twin. See xlambda-core for the adapters and the raw-body rule.

Async

Everything above has an awaitable twin:

from xlambda.media import AsyncMediaClient

async with AsyncMediaClient(api_key=...) as client:
    await client.media.upload(file=data, filename="a.jpg", content_type="image/jpeg")

client.image.transform_url() stays synchronous on both — it builds a string.

Conventions

Arguments are snake_case; responses are the API's own camelCase, returned as plain dicts typed by TypedDict. See xlambda-core.