YouTube video download API in Python: asynchronous jobs

A YouTube video download API lets a Python application submit a source URL and retrieve the resulting media through an asynchronous job. With Tornado API, the basic sequence is create a job, save its ID, poll its status, then consume its delivery URL. Receiving a job ID means the request was accepted; your application still needs to wait for a successful result.

This tutorial implements that sequence using Python's standard library. The script submits one video, waits with a finite polling budget and distinguishes processing failure from a client that stopped waiting. It can also resume an existing job without submitting the video again.

What you need before running the script

Use Python 3.10 or newer, a Tornado API key and one ordinary YouTube video URL that you are authorized to process. Start with a recorded video, rather than a playlist or live stream: those workflows introduce batch or recording options outside this example.

Create or retrieve your key in the dashboard. Check the account's allowance, billing basis and destination settings there. Keep the key in server-side environment configuration. The Python program uses it only for requests to the Tornado API.

The example asks for an MP4 output with a maximum requested resolution of 720p. This sets an output preference; it does not create detail that is absent from the source. Read the create-job reference before adding options such as audio extraction, clips or custom storage.

Understand the two responses

Creating a single video job uses POST https://api.tornadoapi.io/jobs. The JSON response includes job_id. Querying that job uses GET https://api.tornadoapi.io/jobs/{id} with the same x-api-key header.

The status response uses id for the identifier and status for the outcome. When delivery access is available, s3_url contains the file URL. Preserve the returned URL, including its query string. Its name does not mean every delivery destination is Amazon S3.

Keep these two levels separate: an HTTP response describes the API request, while the JSON job status describes the media-processing result. A successful status lookup can return a job that failed.

Submit or resume a job in Python

Save this as youtube_job.py. Set TORNADO_API_KEY using your normal secret configuration. For a new job, set TORNADO_SOURCE_URL to your video URL. To resume, set TORNADO_JOB_ID instead. A resume does not require the source URL.

The program logs the job ID and status, but deliberately does not print your API key or signed delivery link. Its return value is the delivery URL for your next application step.

import json
import os
import time
from urllib.error import HTTPError, URLError
from urllib.parse import quote
from urllib.request import Request, urlopen

BASE_URL = "https://api.tornadoapi.io"
ACTIVE = {"Pending", "Processing"}
UNSUCCESSFUL = {
    "Failed", "Warning", "Skipped", "Cancelled", "CancelledByAdmin"
}


def api_json(method, path, api_key, payload=None, timeout=30):
    body = None if payload is None else json.dumps(payload).encode("utf-8")
    headers = {"x-api-key": api_key, "Accept": "application/json"}
    if body is not None:
        headers["Content-Type"] = "application/json"
    request = Request(BASE_URL + path, data=body, headers=headers, method=method)
    try:
        with urlopen(request, timeout=timeout) as response:
            data = json.load(response)
    except HTTPError as exc:
        code = exc.code
        exc.close()
        raise RuntimeError(f"{method} {path} returned HTTP {code}") from None
    except (URLError, TimeoutError) as exc:
        raise RuntimeError(
            f"{method} {path} had a network error; inspect existing jobs "
            "before submitting again"
        ) from exc
    if not isinstance(data, dict):
        raise RuntimeError("Expected a JSON object from the API")
    return data


def wait_for_delivery(job_id, api_key, max_wait=300):
    deadline = time.monotonic() + max_wait
    path = "/jobs/" + quote(job_id, safe="")
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            raise TimeoutError(
                f"Stopped waiting for {job_id}; resume this job ID later"
            )
        job = api_json("GET", path, api_key, timeout=min(30, remaining))
        status = job.get("status")
        print(f"Job {job_id}: {status}", flush=True)
        if status == "Completed":
            delivery_url = job.get("s3_url")
            if not isinstance(delivery_url, str) or not delivery_url:
                raise RuntimeError(
                    f"Job {job_id} completed without a usable delivery URL; "
                    "inspect its storage destination"
                )
            return delivery_url
        if status in UNSUCCESSFUL:
            raise RuntimeError(
                f"Job {job_id} ended with {status}; inspect error, "
                "error_type and failure_kind in its status response"
            )
        if status not in ACTIVE:
            raise RuntimeError(f"Unexpected job status: {status!r}")
        time.sleep(min(5, max(0, deadline - time.monotonic())))


def main():
    api_key = os.environ["TORNADO_API_KEY"]
    job_id = os.environ.get("TORNADO_JOB_ID")
    if not job_id:
        created = api_json("POST", "/jobs", api_key, {
            "url": os.environ["TORNADO_SOURCE_URL"],
            "format": "mp4",
            "max_resolution": "720",
        })
        job_id = created.get("job_id")
        if not isinstance(job_id, str) or not job_id:
            raise RuntimeError("No single-video job_id; inspect the response")
    print(f"Save this job ID before continuing: {job_id}", flush=True)
    delivery_url = wait_for_delivery(job_id, api_key)
    print("Delivery URL received; ready for your next application step.")
    return delivery_url


if __name__ == "__main__":
    main()

Run it with python3 youtube_job.py. In a larger application, persist the ID in your database as soon as creation succeeds. Associate it with your own customer request or processing record. A console line is useful for this tutorial, but it is not a durable workflow store.

The 300-second waiting budget and five-second interval are example client settings, not processing-time guarantees or recommended account throughput. Each HTTP operation also receives a socket timeout. Python's standard-library socket timeout is not a hard wall-clock limit over every network phase; an application requiring strict overall cancellation should enforce that in its task runner.

Handle every terminal status

The status reference defines these case-sensitive values:

The unsuccessful statuses remain distinct in your records even if they share the same branch in this small script. When present, error provides a message; error_type and failure_kind give additional classification. These fields can be absent, so do not assume every unsuccessful job includes all of them.

An unavailable video, a delivery issue and an application timeout need different responses. A timeout in your Python client does not cancel the remote job or prove that media processing failed. Resume the existing ID to find out what happened.

Avoid creating duplicate work after a timeout

The example does not automatically retry POST /jobs. If the connection drops after the API accepts the request, your client may miss the response even though a job exists. Sending the same request again can create another job.

Inspect the dashboard or your existing job records before resubmitting after an ambiguous creation failure. Once you know the ID, rerun with TORNADO_JOB_ID to continue status checks. For production, persist submission attempts and decide how your application reconciles an uncertain result.

For HTTP authentication or validation errors, correct the request or account configuration first. For rate limits or transient lookup failures, resume status checks with a bounded retry policy appropriate to your workload. Do not translate every unsuccessful outcome into a new media job. The failure and retry guide covers that decision separately.

Pass the delivered file to your application

A delivery URL is an access mechanism, not a permanent identity for the media. Keep the job ID and your destination reference as stable records. Use the returned URL unchanged and avoid putting signed links into public logs, analytics properties or shared tickets.

The API key used in this tutorial belongs on API requests. Do not attach it to a request for the delivery URL. If your next step is transcription, editing or analysis, check that the service accepts the delivered format before starting a larger workload.

A completed job without a usable link needs a storage-access check. Reprocessing the video is not the first recovery step. Review the destination's read access and the storage and retention guide.

Check usage before scaling the Python workflow

Test a few representative sources and inspect their delivered files and account usage. Audio extraction, clipping and low-resolution output can have a billable basis different from file size; consult the usage and costs guide before estimating a larger workload.

Once one video works end to end, add durable job records, a background worker and recovery for interrupted polling. You can then evaluate webhooks for notifications while retaining status lookup as a recovery path. Start with the first-workflow guide if you still need to test the account and storage destination in the dashboard.