transcodely

Official Python SDK for Transcodely — encode video into HLS, DASH and MP4 and write it to your own S3, GCS or R2 bucket, with DRM, signed playback and deterministic output paths. Or let Transcodely host and deliver it.

pip install transcodely

Quick start

import os
from transcodely import Transcodely

with Transcodely(api_key=os.environ["TRANSCODELY_API_KEY"]) as client:
    job = client.jobs.create(
        input_url="https://example.com/source.mp4",
        managed=True,  # write outputs to Transcodely-managed storage
        outputs=[{
            "type": "hls",
            "video": [
                {"codec": "h264", "resolution": "1080p"},
                {"codec": "h264", "resolution": "720p"},
            ],
        }],
        # Optional: encode only a sub-range of the input. Applies job-wide and
        # reduces cost (billing keys off the produced output duration). Omit
        # "end_seconds" (or leave 0) to encode through to the end of the input.
        clip={"start_seconds": 2, "end_seconds": 7},
    )
    print(job.id)  # "job_a1b2c3d4e5f6"

    for event in client.jobs.watch(job.id):
        print(event.job.status, event.job.progress)
        if event.job.status == 4:  # JOB_STATUS_COMPLETED
            break

The simplified-string form ("hls", "h264", "1080p") is what the API actually emits over the wire — the SDK round-trips it transparently to and from the proto enum integers.

Read the output report

Every completed output carries a report of what the produced file actually turned out to be — measured from the encoded file rather than copied from the request — plus the verdict of comparing those measurements against what was asked for.

from transcodely.types import OutputReport

done = client.jobs.get(job.id)

for output in done.outputs:
    if not output.HasField("report"):
        continue  # not measured — never "nothing wrong"

    report: OutputReport = output.report
    print(
        output.id,
        report.video.codec,
        f"{report.video.width}x{report.video.height}",
        f"{report.duration_seconds}s",
    )

    if not report.verdict.matches_request:
        for m in report.verdict.mismatches:
            print(f"  {m.field}: asked for {m.expected}, got {m.actual}")

Check output.HasField("report") rather than truthiness: an absent report means "not measured", which is not the same as a report that found nothing wrong. Branch on m.field — it comes from a fixed vocabulary (video.codec, video.resolution, duration_seconds, …) — rather than on the values beside it. For an ABR ladder the facts describe the highest-resolution rendition, the same one the verdict judges; per-rendition detail stays in variant_results.

An output encoded with per-title content-aware analysis also carries report.content_aware: the VMAF target the search aimed at, the score it reached on its samples, and the CRF it chose. It describes the SEARCH, not the delivered file — vmaf_achieved scores short samples taken before the real encode, which is never scored itself. Check report.HasField("content_aware"); an ordinary output has none.

AI captions

Add auto-generated captions to any output with a generate subtitle track. Leave language empty (or set "auto") to auto-detect the spoken language, or pass an ISO 639-2 code to force one. A per-job fee is metered by source minute and surfaced on job.fees; produced captions appear on job.subtitle_results with auto_generated=True.

# Generate captions while transcoding a new source.
client.jobs.create(
    input_url="https://example.com/source.mp4",
    managed=True,
    outputs=[{
        "type": "hls",
        "video": [{"codec": "h264", "resolution": "1080p"}],
        "subtitle_tracks": [{"operation": "generate", "language": "auto"}],
    }],
)

# Retro-caption a video you've already hosted: reference it by input_video_id and
# request a single captions-only output (no video encode).
job = client.jobs.create(
    input_video_id="vid_a1b2c3d4e5f6g7",
    outputs=[{"subtitle_tracks": [{"operation": "generate"}]}],
)

for result in job.subtitle_results:
    print(result.language, result.auto_generated, result.url)
for fee in job.fees:
    print(fee.fee_type, fee.amount, fee.currency)  # "captions" 0.51 "eur"

Authentication

client = Transcodely(api_key=os.environ["TRANSCODELY_API_KEY"])

API keys are opaque ak_-prefixed secrets — pass the full value shown once at creation time.

Resources

client.jobs            # create / get / list / cancel / confirm / watch
client.videos          # upload helpers, multipart, create_from_url, get / list / update / delete / watch, get_stats / list_top_videos
client.presets         # create / get / get_by_slug / list / update / duplicate / archive
client.origins         # create / get / list / update / validate / archive
client.ingest_rules    # create / get / list / update / delete / list_events / test / replay_event
client.apps            # create / get / list / update / archive / enable_hosting
client.api_keys        # create / get / list / revoke
client.organizations   # create / get / list / update / check_slug
client.billing         # list_invoices / retrieve / retrieve_upcoming / retrieve_profile / create_portal_session
                       #   + retrieve_budget / set_budget / clear_budget
                       #   + retrieve_outstanding_balance / settle_outstanding_balance
client.memberships     # list / get / update_role / remove
client.users           # get_me / get / list / update_me
client.health          # check
client.webhook_endpoints  # create / retrieve / update / delete / list / rotate_secret / send_test / list_deliveries / get_health
client.events          # retrieve / list / resend
client.webhooks        # construct_event / verify_signature (signature-verification helpers)

Videos

One-call ingest from a URL

create_from_url skips the upload dance entirely: give it a publicly-reachable http(s) URL and the worker fetches it at transcode time. Requires an app with managed hosting enabled (client.apps.enable_hosting(...)).

video = client.videos.create_from_url(
    app_id="app_default000",
    url="https://example.com/source.mp4",
    title="Product demo",
)
print(video.id, video.status)  # "processing" — no upload step

for event in client.videos.watch(video.id):
    if event.video.status == "ready":
        print(event.video.playback_url, event.video.embed_url)
        break

No video.uploaded event fires for URL ingest (no bytes are uploaded to the API) — subscribe to video.ready / video.failed instead to know when it's playable.

Playback analytics

get_stats returns a single video's playback stats — plays, watch time, and unique viewers — aggregated per UTC day, plus range totals. list_top_videos ranks an app's videos by plays over a date range. Both accept optional start_date / end_date (YYYY-MM-DD, inclusive); stats are best-effort and roll up hourly, so recent activity can lag by up to an hour.

stats = client.videos.get_stats(
    video_id="vid_abc123def456",
    start_date="2026-07-01",
    end_date="2026-07-16",
)
for day in stats.daily:
    print(day.date, day.plays, day.watch_seconds, day.unique_viewers)
print("range totals:", stats.totals.plays, stats.totals.watch_seconds)

top = client.videos.list_top_videos(app_id="app_default000", limit=10)
for video in top.items:
    print(video.video_id, video.title, video.plays, video.watch_seconds)

Origins

An origin tells Transcodely where to read source media from and where to write outputs. Every origin belongs to a single provider; pass exactly one provider-config field (s3, gcs, http, or r2) on create.

Create an S3 origin

origin = client.origins.create(
    name="Production S3",
    permissions=["read", "write"],
    s3={
        "bucket": "my-bucket",
        "region": "us-east-1",
        "credentials": {
            "access_key_id": os.environ["S3_ACCESS_KEY"],
            "secret_access_key": os.environ["S3_SECRET_KEY"],
        },
        # "endpoint": "https://s3.custom.example.com",  # for MinIO, Wasabi, etc.
    },
)

Create a GCS origin

origin = client.origins.create(
    name="Production GCS",
    permissions=["read", "write"],
    gcs={
        "bucket": "my-gcs-bucket",
        "credentials": {
            "service_account_json": os.environ["GCS_SERVICE_ACCOUNT_JSON"],
        },
    },
)

Create an HTTP origin

origin = client.origins.create(
    name="Public CDN",
    permissions=["read"],  # HTTP origins are read-only
    http={
        "base_url": "https://media.example.com",
        "credentials": {
            "headers": {"Authorization": f"Bearer {os.environ['MEDIA_TOKEN']}"},
        },
    },
)

Create an R2 origin

R2 supports two forms. With account_id (32-char hex) the endpoint is derived for you, optionally with a data-residency jurisdiction:

origin = client.origins.create(
    name="Production R2",
    permissions=["read", "write"],
    r2={
        "bucket": "media",
        "account_id": os.environ["R2_ACCOUNT_ID"],
        "jurisdiction": "default",  # or "eu", "fedramp"
        "credentials": {
            "access_key_id": os.environ["R2_ACCESS_KEY"],
            "secret_access_key": os.environ["R2_SECRET_KEY"],
        },
    },
)

Or, with an explicit endpoint (custom domain bound to a bucket, or a jurisdiction not yet enumerated):

r2 = {
    "bucket": "media",
    "endpoint": "https://media.example.com",
    "credentials": {
        "access_key_id": os.environ["R2_ACCESS_KEY"],
        "secret_access_key": os.environ["R2_SECRET_KEY"],
    },
}

Provide either account_id or endpoint, never both. jurisdiction only applies when account_id is set.

Ingest rules

An ingest rule is a standing instruction on one readable origin: when an object matching these filters lands, create this job for it. Your storage provider posts its object-created events to the rule's endpoint, and no server of yours is in the path. Amazon S3 via SNS, Google Cloud Storage via a Pub/Sub push subscription, Supabase Storage via a database webhook, and a generic shape for anything else are all recognised from the payload.

created = client.ingest_rules.create(
    origin_id="ori_a1b2c3d4e5f6",
    name="Watch uploads/",
    filters={
        "prefix": "uploads/",
        "suffixes": [".mp4", ".mov"],
        "min_bytes": 1024,  # ignore the zero-byte placeholder some clients write first
    },
    action={
        "outputs": [{"preset": "web_1080p_standard"}],
        "managed": True,  # host and deliver the result; use output_origin_id for your own bucket
        "priority": "standard",
    },
)

print("point your bucket notifications at", created.rule.endpoint_url)
print("secret (shown once):", created.secret)

created.secret is the only time the inbound secret is readable — store it wherever the event sender will read it from. A later get returns just secret_prefix and secret_hint. Lost it? update with rotate_secret=True issues a new one, and the previous one keeps working for 24 hours so the sender can be changed without dropping an event.

Every delivery is recorded, whether or not it became a job:

for event in client.ingest_rules.list_events(rule_id=created.rule.id).auto_paging_iter():
    print(event.id, event.object_key, event.status, event.reason)

A skipped event names why in reason — filter_prefix, filter_suffix, filter_content_type, filter_size, bucket_mismatch, rule_disabled, or duplicate. A failed one carries the API error code that refused the job, such as limit_exceeded. Narrow the log to either with the simplified status string: list_events(status="skipped").

Deduplication is permanent: an object is identified by (rule, bucket, key, etag), so re-sending the event or re-uploading the same bytes produces nothing. To give an object another pass — one that arrived while the rule was paused, or was refused while the account was over its cap — replay it:

replayed = client.ingest_rules.replay_event("sev_a1b2c3d4e5f6g7")
print(replayed.id, "is back in", StorageEventStatus.Name(replayed.status))

Only skipped and failed events can be replayed. When an update switches a paused rule back on, the response reports how large that backlog is in events_skipped_while_disabled.

Before wiring the provider up, dry-run a key against the rule with client.ingest_rules.test(...): it reports whether the filters match and, when they do, the exact job request the rule would submit. Nothing is stored.

Billing

Billing settles a whole organization, so client.billing needs a dashboard session token for an organization owner plus the organization it is for — an API key is scoped to one app and is rejected outright:

client = Transcodely("tk_session_token", organization_id="org_f6g7h8i9j0")

client.billing.retrieve_upcoming()  # what the current period has accrued
client.billing.list_invoices(limit=12)

Three numbers live here and they are not interchangeable:

Budgets notify; limits block:

client.billing.set_budget(500.0)  # emails at 50% / 80% / 100%, restricts nothing
client.billing.clear_budget()  # turns the alerts off

The outstanding balance is the platform's own exposure control. Reminder emails go out at 80%, 100%, 125%, 150% and 175% of threshold_cents and restrict nothing; only at hard_stop_cents (twice the threshold) are new jobs refused with code outstanding_balance_exceeded — queued and running work still finishes and videos keep playing. Paying lifts the block on the next request:

balance = client.billing.retrieve_outstanding_balance()
if balance.blocked and balance.settlement_available:
    result = client.billing.settle_outstanding_balance()
    print(result.settlement.invoice_id, result.balance.outstanding_cents)  # → 0

Check settlement_available before offering a "pay now" action: settlement raises PreconditionError with code settlement_unavailable where the rail is switched off, and nothing_outstanding when there is nothing to pay. Job creates refused for money reasons carry billing_past_due (a statement is unpaid) or outstanding_balance_exceeded (too much unbilled usage) — neither clears by retrying, so back-off is the wrong response to both.

Errors

All exceptions inherit from TranscodelyError:

from transcodely import (
    Transcodely,
    TranscodelyError,
    InvalidRequestError,
    NotFoundError,
    RateLimitError,
)

try:
    client.jobs.create(input_url=..., outputs=[...])
except InvalidRequestError as err:
    for v in err.errors:
        print(f"{v.field}: {v.description}")
except RateLimitError as err:
    time.sleep((err.retry_after_ms or 1000) / 1000)
except TranscodelyError as err:
    print(f"[{err.request_id}] {err.code}: {err}")

Every error carries request_id, code, http_status, and raw for debugging.

Webhook verification raises WebhookError (or a subclass) — see Webhooks.

Pagination

# One page
page = client.jobs.list(limit=50)
print(page.items, page.next_cursor)

# All items, automatically across pages
for job in client.jobs.list(limit=50).auto_paging_iter():
    print(job.id)

Idempotency

jobs.create accepts idempotency_key. If you don't pass one, the SDK generates a UUID v4 so retries are safe by default. For cross-process safety, pass your own:

client.jobs.create(
    input_url="...",
    outputs=[...],
    idempotency_key="create-job-for-asset-12345",
)

For all other write methods, the SDK ships an Idempotency-Key HTTP header automatically.

Streaming watch

for event in client.jobs.watch(job.id):
    print(event.event, event.job.status, event.job.progress)

The SDK auto-reconnects on transient network failures — Watch is read-only and re-emits a SNAPSHOT on every reconnect. HEARTBEAT events are filtered by default.

Webhooks

Verify a signed delivery and get back a typed Event. Pass the raw request body (bytes or str — do not re-serialize) and the Transcodely-Signature header:

from transcodely import construct_event, WebhookSignatureError

try:
    event = construct_event(
        request.body,                       # raw bytes/str, exactly as received
        request.headers["transcodely-signature"],
        "whsec_...",                        # your endpoint's signing secret
    )
except WebhookSignatureError:
    return Response(status_code=400)

# `event.data` is the decoded resource (a Job, JobOutput, Video, or App).
if event.type == "job.succeeded":
    print("job done:", event.data.id)
elif event.type == "video.uploaded":
    print("new video:", event.data.id)

construct_event also accepts a list of secrets so deliveries keep verifying during a secret rotation's overlap window:

event = construct_event(body, sig_header, ["whsec_previous", "whsec_current"])

Tuning and errors:

• tolerance (default 300 seconds) bounds clock skew / replay; widen or narrow it per call.
• WebhookSignatureError — header malformed or no signature matched.
• WebhookTimestampError — timestamp outside the tolerance window.
• WebhookPayloadError — body isn't valid JSON or doesn't match the event envelope.

All three inherit from WebhookError. client.webhooks.construct_event(...) is an alias for the module-level function.

Manage endpoints and replay events via the API:

endpoint = client.webhook_endpoints.create(
    app_id="app_123",
    url="https://example.com/hooks/transcodely",
    enabled_events=["job.succeeded", "job.failed"],   # or ["*"] for all
)
print(endpoint.secret)   # shown only on create + rotate_secret — store it now

# The same typed Event, fetched from the API instead of an HTTP delivery:
event = client.events.retrieve("evt_123")
for event in client.events.list(app_id="app_123").auto_paging_iter():
    print(event.type, event.id)

client.events.resend("evt_123")   # re-queue delivery to all subscribed endpoints

The 17 event types are job.created, job.succeeded, job.failed, job.canceled, job.progress, output.created, output.ready, output.failed, output.progress, video.uploaded, video.ready, video.failed, video.deleted, app.created, app.updated, app.spend_limit_warning, and app.spend_limit_exceeded. Subscribe to "*" to receive all of them (including ones added later). An unrecognized future type still verifies; its event.data is left as a plain dict. The two app.spend_limit_* events also carry a plain dict (app_id, period_start, period_end, limit_eur, spent_eur, threshold_pct, currency), not a resource snapshot.

Configuration

Transcodely(
    api_key,                    # required
    base_url=None,              # default: https://api.transcodely.com
    timeout=30.0,               # seconds
    max_retries=3,
    api_version=None,           # override the pinned API version
    default_headers=None,       # dict, sent on every request
    http_client=None,           # custom httpx.Client
    logger=None,                # callable(LogEvent)
)

Request IDs

client.jobs.get("job_x")
print(client.last_request_id)  # "req_*"

Errors also carry the request ID via err.request_id for log correlation.

Versioning

The SDK follows semver, starting at 0.1.0. Breaking changes are allowed on minor bumps until 1.0.0. Each release pins a specific calendar-versioned API (Transcodely.API_VERSION) and sends Transcodely-Version on every request.

License

MIT.