Skip to content
Sign in Get an API key
On this page

DocsIntegrations

Python with FastAPI

A FastAPI app with async httpx calls that creates jobs from JSON, takes uploads straight from the browser, verifies webhooks and hands out downloads.

What you build#

A FastAPI app that turns an uploaded file into an MP4, or into an MP4, a poster and a GIF preview in one job. The browser sends the file straight to Videxo with a one-time URL, so it never passes through your server. Your server keeps the API key, creates the job, verifies the webhook and sends each download to a fresh signed link. Every call to Videxo sits in one module, app/videxo.py, with plain HTTP (httpx).

The whole project: videxo-python-fastapi.zip, or browse it in the examples repository (opens in a new tab).

Before you start#

  • An API key from the dashboard's API keys page, in VIDEXO_API_KEY.
  • The webhook signing secret from the dashboard's Webhooks page, in VIDEXO_WEBHOOK_SECRET.
  • Python 3.10 or newer.

Both values stay on the server, in environment variables. Never put them in code that runs in a browser.

Install#

bash
python -m venv .venv
.venv/bin/pip install -r requirements.txt

requirements.txt holds fastapi, uvicorn, httpx and standardwebhooks (opens in a new tab), which verifies the webhook's signature.

Create a job#

The JSON bodies sit in requests/, next to the code: one MP4 through POST /v1/transcode, and an MP4, a poster and a GIF preview in one job through POST /v1/jobs. The functions put the uploaded file in as the input and send an Idempotency-Key:

examples/python-fastapi/app/videxo.py · plain HTTP
async def transcode(input_url: str, idempotency_key: str) -> dict[str, Any]:
    """One MP4 from one input: POST /v1/transcode with the body in requests/transcode.json."""
    body = {**TRANSCODE_BODY, "input": input_url, **_webhook()}
    return await _json("POST", "/v1/transcode", body, idempotency_key)


async def create_job(input_url: str, idempotency_key: str) -> dict[str, Any]:
    """Several outputs from one input in one job: POST /v1/jobs with the body in requests/jobs.json."""
    body = {**JOBS_BODY, "inputs": {"talk": input_url}, **_webhook()}
    return await _json("POST", "/v1/jobs", body, idempotency_key)

Every function goes through one _request helper. An error response is problem details; it becomes a VidexoProblem with the code, the detail and the field errors:

examples/python-fastapi/app/videxo.py · plain HTTP
class VidexoProblem(Exception):
    """An error answer of the API: RFC 9457 problem details. Branch on `code`; show `detail` to people."""

    def __init__(self, problem: dict[str, Any]) -> None:
        super().__init__(f"{problem.get('code')}: {problem.get('detail')}")
        self.problem = problem
        self.status: int = problem.get("status", 500)


_client = httpx.AsyncClient(base_url=API_URL, timeout=60)


async def _request(method: str, path: str, body: Any = None, idempotency_key: str | None = None) -> httpx.Response:
    api_key = os.environ.get("VIDEXO_API_KEY")
    if not api_key:
        raise RuntimeError("Set VIDEXO_API_KEY in the server environment.")
    headers = {"Authorization": f"Bearer {api_key}"}
    if idempotency_key:
        headers["Idempotency-Key"] = idempotency_key
    # httpx does not follow redirects unless asked: the download route reads the Location itself.
    response = await _client.request(method, path, json=body, headers=headers)
    if response.status_code >= 400:
        try:
            problem = response.json()
        except ValueError:
            problem = None
        if not isinstance(problem, dict) or "code" not in problem:
            # Not problem details: a proxy or the network answered, not the API.
            problem = {
                "type": "about:blank",
                "title": "The API answered with an error.",
                "status": response.status_code,
                "detail": f"HTTP {response.status_code}",
                "code": "http_error",
            }
        raise VidexoProblem(problem)
    return response


async def _json(method: str, path: str, body: Any = None, idempotency_key: str | None = None) -> dict[str, Any]:
    data: dict[str, Any] = (await _request(method, path, body, idempotency_key)).json()
    return data

The route takes the upload's input_url from the page and derives the idempotency key from the upload, so a retried request returns the same job:

examples/python-fastapi/app/main.py · plain HTTP
UPLOAD = re.compile(r"^videxo://uploads/(up_[a-f0-9]{24})$")


@app.post("/api/jobs", status_code=202)
async def create_job(request: Request) -> dict[str, Any]:
    body = await request.json()
    input_url = body.get("input_url")
    # Accept only uploads, so nobody can make this server fetch URLs of their choice.
    match = UPLOAD.match(input_url) if isinstance(input_url, str) else None
    if not match:
        raise bad_request("input_url must be the input_url of an upload.")
    upload = match.group(1)
    # The same key for the same upload and kind: a retried request answers the same job, never a second one.
    if body.get("kind") == "jobs":
        return await videxo.create_job(input_url, f"jobs-{upload}")
    return await videxo.transcode(input_url, f"transcode-{upload}")

An exception handler passes Videxo's problem details to the page as they came, so the page can show what was wrong:

examples/python-fastapi/app/main.py · plain HTTP
@app.exception_handler(videxo.VidexoProblem)
async def videxo_problem(_request: Request, error: videxo.VidexoProblem) -> JSONResponse:
    # Videxo's problem details go to the page as they came, so it can show `detail` and the field errors.
    return JSONResponse(error.problem, status_code=error.status, media_type="application/problem+json")

Receive the webhook#

When the job ends, Videxo sends a signed POST (Webhooks). The route reads the raw body exactly as it arrived, refusing one over 2 MB with 413 (Content-Length first, then it stops reading past the limit), and checks it:

examples/python-fastapi/app/videxo.py · plain HTTP
def verify_webhook(body: bytes, headers: dict[str, str]) -> dict[str, Any]:
    """The event when the request is a webhook Videxo signed, in the last five minutes; it raises
    WebhookVerificationError otherwise. `body` is the raw body exactly as received: parsed and
    serialized again, it no longer matches the signature."""
    secret = os.environ.get("VIDEXO_WEBHOOK_SECRET")
    if not secret:
        raise RuntimeError("Set VIDEXO_WEBHOOK_SECRET in the server environment.")
    event: dict[str, Any] = Webhook(secret).verify(body, headers)
    return event
examples/python-fastapi/app/main.py · plain HTTP
# A webhook body is the whole job: up to several hundred KB, so 2 MB leaves room.
MAX_WEBHOOK_BYTES = 2 * 1024 * 1024


async def read_limited(request: Request, limit: int) -> bytes | None:
    """The raw body, or None when it is longer than `limit` bytes; never reads more than `limit` + 1."""
    length = request.headers.get("content-length")
    if length is not None and length.isdigit() and int(length) > limit:
        return None
    body = bytearray()
    async for chunk in request.stream():
        body += chunk
        if len(body) > limit:
            return None
    return bytes(body)


@app.post("/webhooks/videxo")
async def webhook(request: Request) -> Response:
    body = await read_limited(request, MAX_WEBHOOK_BYTES)
    if body is None:
        return Response(status_code=413)
    try:
        event = videxo.verify_webhook(body, dict(request.headers))
    # The library raises ValueError, not its own error, for a webhook-signature that is not "v1,<base64>".
    except (WebhookVerificationError, ValueError):
        return Response(status_code=400)
    webhook_id = request.headers["webhook-id"]
    if webhook_id not in handled:
        handled.add(webhook_id)
        on_job_ended(event)
    # Answer 2xx within seconds; do slow work in a queue of your own (FastAPI's BackgroundTasks, a worker).
    return Response(status_code=204)

verify rejects a wrong signature and a timestamp more than five minutes away. For a webhook-signature header that is not v1, and base64 it raises ValueError instead of its own error, so the route catches both. A delivery can arrive more than once with the same webhook-id; keep the ids you have handled in a unique column and skip one you have seen.

Upload a file#

The server creates the upload with the API key and gives the browser only its one-time URL:

examples/python-fastapi/app/videxo.py · plain HTTP
async def create_upload(filename: str, size_bytes: int, content_type: str) -> dict[str, Any]:
    """A one-time upload URL for a file the browser sends straight to Videxo; the API key stays here."""
    body = {"filename": filename, "size_bytes": size_bytes, "content_type": content_type}
    return await _json("POST", "/v1/uploads", body)
examples/python-fastapi/app/main.py · plain HTTP
@app.post("/api/uploads", status_code=201)
async def create_upload(request: Request) -> dict[str, str]:
    body = await request.json()
    upload = await videxo.create_upload(body.get("filename"), body.get("size_bytes"), body.get("content_type"))
    # Only the one-time URL goes to the browser; it holds no API key.
    return {"input_url": upload["input_url"], "upload_url": upload["upload_url"]}

The page then sends the file with PUT to upload_url (no API key; the URL takes one PUT) and asks the server for the job with the upload's input_url, videxo://uploads/<id>. See Files for the size limit and how long an upload lasts.

Download the outputs#

Signed links expire within an hour, so the download route asks for a fresh one each time: GET /v1/jobs/{id}/outputs/{alias} returns a redirect to it, which httpx does not follow, and the route sends the browser on.

examples/python-fastapi/app/videxo.py · plain HTTP
async def output_url(job_id: str, alias: str) -> str:
    """A fresh signed link to one output, from the redirect of GET /v1/jobs/{id}/outputs/{alias}."""
    response = await _request("GET", f"/v1/jobs/{job_id}/outputs/{alias}")
    return response.headers["location"]
examples/python-fastapi/app/main.py · plain HTTP
@app.get("/api/jobs/{job_id}/outputs/{alias}")
async def download(job_id: str, alias: str) -> RedirectResponse:
    # A fresh link each time: signed links expire within an hour.
    return RedirectResponse(await videxo.output_url(job_id, alias), status_code=302)

A job's outputs[].download_url is also a signed link, valid until download_url_expires_at.

Run it#

bash
cp .env.example .env    # fill in VIDEXO_API_KEY and VIDEXO_WEBHOOK_SECRET
.venv/bin/uvicorn app.main:app --env-file .env --port 3000

Open http://localhost:3000, pick a file and choose the job. The page follows the job by reading it every 2 seconds, then lists the outputs as download links.

Webhooks need a public https URL. In development run a tunnel (for example cloudflared tunnel --url http://localhost:3000) and set VIDEXO_WEBHOOK_URL to its address plus /webhooks/videxo; each job then carries it as its webhook_url. Or set the address once as your account's endpoint on the Webhooks page.

Next steps#