Python

pip install requests

# videotranscode.py
import os
import requests
 
BASE = "https://videotranscode.cloud/api/v1"
 
 
class VideoTranscodeError(Exception):
    def __init__(self, status, body):
        self.status = status
        self.body = body or {}
        message = self.body.get("message") or self.body.get("error") or f"Request failed with {status}"
        super().__init__(message)
 
 
class Client:
    def __init__(self, api_key=None, timeout=30):
        self.api_key = api_key or os.environ["VT_API_KEY"]
        self.timeout = timeout
        self.session = requests.Session()
        self.session.headers.update({
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        })
 
    def _request(self, method, path, **kwargs):
        response = self.session.request(method, f"{BASE}{path}", timeout=self.timeout, **kwargs)
 
        try:
            body = response.json()
        except ValueError:
            body = None
 
        if not response.ok:
            raise VideoTranscodeError(response.status_code, body)
 
        return body
 
    def create_job(self, input_url, preset, metadata=None):
        payload = {"inputUrl": input_url, "preset": preset}
        if metadata:
            payload["metadata"] = metadata
        return self._request("POST", "/jobs", json=payload)
 
    def get_job(self, job_id):
        return self._request("GET", f"/jobs/{job_id}")
 
    def list_jobs(self, **params):
        return self._request("GET", "/jobs", params=params)
 
    def cancel_job(self, job_id):
        return self._request("DELETE", f"/jobs/{job_id}")
 
    def download_url(self, job_id):
        return self._request("GET", f"/jobs/{job_id}/download-url")
 
    def list_presets(self):
        return self._request("GET", "/presets")

from videotranscode import Client, VideoTranscodeError
 
client = Client()
 
try:
    job = client.create_job(
        "https://cdn.example.com/master.mov",
        "hls_abr_264_standard",
    )
    print(f"Queued {job['id']}, {job['user']['remainingJobs']} jobs left this month")
 
except VideoTranscodeError as error:
    if error.status == 403:
        # The body names what you can use instead.
        print("Not on your plan. Available:", ", ".join(error.body["availablePresets"]))
    elif error.status == 429:
        usage = error.body["usage"]
        print(f"Quota used: {usage['current']} of {usage['limit']}")
    elif error.status == 400:
        print(error.body.get("details") or error.body["message"])
    else:
        raise

import time
 
TERMINAL = {"completed", "failed", "cancelled"}
 
 
def wait_for_job(client, job_id, interval=12, timeout=45 * 60):
    deadline = time.monotonic() + timeout
 
    while time.monotonic() < deadline:
        job = client.get_job(job_id)
        if job["status"] in TERMINAL:
            return job
        time.sleep(interval)
 
    raise TimeoutError(f"Job {job_id} did not finish within {timeout} seconds")

import requests
from videotranscode import Client
 
client = Client()
 
job = client.create_job("https://cdn.example.com/master.mov", "mp4_264_720p")
print(f"Created {job['id']}")
 
finished = wait_for_job(client, job["id"])
 
if finished["status"] != "completed":
    raise RuntimeError(f"Job {finished['id']} {finished['status']}: {finished['errorMessage']}")
 
# The signature is in the query string, so no auth header here.
with requests.get(finished["outputUrl"], stream=True, timeout=300) as response:
    response.raise_for_status()
    with open("output.mp4", "wb") as handle:
        for chunk in response.iter_content(chunk_size=1024 * 1024):
            handle.write(chunk)
 
print(f"Saved output.mp4, link was valid until {finished['expiration']}")

import hashlib
import hmac
import json
import os
from flask import Flask, request, jsonify
 
app = Flask(__name__)
SECRET = os.environ["VT_WEBHOOK_SECRET"]
 
 
def signature_is_valid(raw_body, header):
    if not header:
        return False
 
    expected = hmac.new(SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
 
    # Constant time, so the check cannot be probed byte by byte.
    return hmac.compare_digest(expected, header)
 
 
@app.post("/hooks/video")
def receive():
    raw = request.get_data()
 
    if not signature_is_valid(raw, request.headers.get("X-VideoTranscoder-Signature")):
        return jsonify({"error": "Invalid signature"}), 401
 
    payload = json.loads(raw)
 
    # Queue the work; delivery times out after 30 seconds.
    if payload["status"] == "completed":
        enqueue_archive(payload["jobId"], payload["outputUrl"])
    elif payload["status"] == "failed":
        enqueue_failure(payload["jobId"], payload.get("error"))
 
    return jsonify({"received": True}), 200