Video generation

import os, time, httpx

http = httpx.Client(
    base_url="https://api.infery.ai/v1",
    headers={"Authorization": f"Bearer {os.environ['INFERY_API_KEY']}"},
    timeout=60,
)

def generate_video(prompt: str, model: str = "veo-3.1-fast") -> dict:
    job = http.post(
        "/videos/generations",
        json={
            "model": model,
            "prompt": prompt,
            "duration": 4,
            "resolution": "720p",
            "aspect_ratio": "16:9",
        },
    ).json()
    job_id = job["id"]

    while True:
        status = http.get(f"/videos/generations/{job_id}").json()
        if status["status"] == "completed":
            return status["result"]
        if status["status"] == "failed":
            raise RuntimeError(status.get("error", "video generation failed"))
        time.sleep(5)

async function* watchJob(jobId: string) {
  while (true) {
    const r = await fetch(`https://api.infery.ai/v1/videos/generations/${jobId}`, {
      headers: { Authorization: `Bearer ${process.env.INFERY_API_KEY}` },
    }).then((r) => r.json());

    yield r;

    if (r.status === 'completed' || r.status === 'failed') return;
    await new Promise((res) => setTimeout(res, 5000));
  }
}

for await (const update of watchJob(jobId)) {
  ui.setProgress(update.progress ?? 0);
  if (update.status === 'completed') ui.showVideo(update.result.url);
}

job = http.post("/videos/generations", json={
    "model": "veo-3.1",
    "prompt": "Slow camera dolly-in, gentle wind in the leaves",
    "image_url": "https://cdn.example.com/forest.jpg",
    "duration": 8,
    "aspect_ratio": "16:9",
}).json()

import httpx
from openai import OpenAI

oai = OpenAI(api_key=os.environ["INFERY_API_KEY"], base_url="https://api.infery.ai/v1")

result = generate_video("A vibrant coral reef, slow cinematic dolly-in")
mp4 = httpx.get(result["url"], follow_redirects=True).content

f = oai.files.create(file=("hero.mp4", mp4), purpose="user_data")
# f.id is now a stable handle in your workspace storage