Exemplos em Python - Shappire API

Instalação

pip install requests
# ou
pip install httpx

Cliente com requests

import os
import time
import requests

API_KEY = os.environ["SHAPPIRE_API_KEY"]
BASE = "https://api.shappire.tools/v1"
SESSION = requests.Session()
SESSION.headers.update({"X-API-Key": API_KEY})

def resolve(url: str) -> dict:
    r = SESSION.post(f"{BASE}/media/resolve", json={"url": url}, timeout=30)
    r.raise_for_status()
    return r.json()["data"]

def enqueue(media_id: str, format: str) -> str:
    r = SESSION.post(
        f"{BASE}/media/download",
        json={"media_id": media_id, "format": format},
        timeout=30,
    )
    r.raise_for_status()
    return r.json()["data"]["job_id"]

def wait_job(job_id: str, interval: float = 3.0) -> dict:
    while True:
        r = SESSION.get(f"{BASE}/jobs/{job_id}", timeout=30)
        r.raise_for_status()
        data = r.json()["data"]
        if data["status"] == "completed":
            return data
        if data["status"] == "failed":
            raise RuntimeError(data.get("error"))
        time.sleep(interval)

Fluxo completo

def download_from_url(url: str) -> str:
    media = resolve(url)
    job_id = enqueue(media["id"], "video_best")
    job = wait_job(job_id)
    return job["result"]["download_url"]

Com httpx (async)

import httpx
import asyncio

async def resolve_async(url: str) -> dict:
    async with httpx.AsyncClient(
        base_url="https://api.shappire.tools/v1",
        headers={"X-API-Key": API_KEY},
        timeout=30,
    ) as client:
        r = await client.post("/media/resolve", json={"url": url})
        r.raise_for_status()
        return r.json()["data"]

FastAPI

from fastapi import FastAPI, HTTPException

app = FastAPI()

@app.post("/download")
def create_download(url: str):
    try:
        return {"download_url": download_from_url(url)}
    except requests.HTTPError as exc:
        raise HTTPException(status_code=exc.response.status_code, detail=str(exc))

Django / Celery

Enfileire o download na API e deixe o worker Celery fazer polling:

@celery_app.task
def process_download(url: str):
    media = resolve(url)
    job_id = enqueue(media["id"], "audio_best")
    job = wait_job(job_id)
    save_to_s3(job["result"]["download_url"])

⌘I