async_client.py

"""
Seedance 2.0 – Async Client (httpx)
=====================================
Fully async implementation using httpx for high-throughput applications.
Ideal for FastAPI backends, async pipelines, and production services.

Requirements: pip install httpx

Get your API key: https://www.atlascloud.ai/seedance-2?utm_source=github&utm_campaign=free-seedance-2-api
"""

import asyncio
import os
import httpx

API_KEY = os.environ.get("ATLASCLOUD_API_KEY", "your_api_key_here")
BASE_URL = "https://api.atlascloud.ai"


class SeedanceClient:
    """Async client for the Seedance 2.0 API on Atlas Cloud."""

    def __init__(self, api_key: str = API_KEY):
        self.api_key = api_key
        self.base_url = BASE_URL
        self._client: httpx.AsyncClient | None = None

    @property
    def headers(self) -> dict:
        return {
            "Content-Type": "application/json",
            "Authorization": f"Bearer {self.api_key}",
        }

    async def __aenter__(self):
        self._client = httpx.AsyncClient(timeout=60)
        return self

    async def __aexit__(self, *args):
        if self._client:
            await self._client.aclose()

    async def submit(
        self,
        prompt: str,
        model: str = "bytedance/seedance-2.0/text-to-video",
        width: int = 1280,
        height: int = 720,
        duration: int = 5,
        **kwargs,
    ) -> str:
        """Submit a generation job. Returns prediction_id."""
        input_params = {
            "prompt": prompt,
            "width": width,
            "height": height,
            "duration": duration,
            **kwargs,
        }
        response = await self._client.post(
            f"{self.base_url}/api/v1/model/generateVideo",
            headers=self.headers,
            json={"model": model, "input": input_params},
        )
        response.raise_for_status()
        return response.json()["data"]["id"]

    async def poll(self, prediction_id: str, interval: float = 3.0) -> str:
        """Poll until the job completes. Returns video URL."""
        url = f"{self.base_url}/api/v1/model/prediction/{prediction_id}"
        headers = {"Authorization": f"Bearer {self.api_key}"}
        current_interval = interval

        while True:
            response = await self._client.get(url, headers=headers)
            response.raise_for_status()
            data = response.json()["data"]

            if data["status"] in ("completed", "succeeded"):
                return data["outputs"][0]
            if data["status"] == "failed":
                raise RuntimeError(data.get("error", "Generation failed"))

            await asyncio.sleep(current_interval)
            current_interval = min(current_interval * 1.3, 10)

    async def generate(self, prompt: str, **kwargs) -> str:
        """Submit and poll in one call. Returns video URL."""
        prediction_id = await self.submit(prompt, **kwargs)
        return await self.poll(prediction_id)

    async def upload_media(self, file_path: str) -> str:
        """Upload a local file. Returns temporary URL."""
        with open(file_path, "rb") as f:
            response = await self._client.post(
                f"{self.base_url}/api/v1/model/uploadMedia",
                headers={"Authorization": f"Bearer {self.api_key}"},
                files={"file": f},
            )
        response.raise_for_status()
        return response.json()["data"]["url"]


async def main():
    async with SeedanceClient() as client:
        # Single video
        url = await client.generate(
            "A time-lapse of storm clouds forming over a city skyline, cinematic wide angle",
            width=1920,
            height=1080,
            duration=8,
        )
        print(f"Video: {url}")

        # Parallel generation
        prompts = [
            "Ocean waves crashing on rocky cliffs at sunset, aerial drone shot",
            "Cherry blossoms falling in slow motion, Japanese garden, serene",
            "Northern lights dancing over a frozen lake, long exposure style",
        ]

        tasks = [client.generate(p, duration=5) for p in prompts]
        urls = await asyncio.gather(*tasks)

        for i, url in enumerate(urls):
            print(f"Video {i+1}: {url}")


if __name__ == "__main__":
    asyncio.run(main())