Python SDK Guide

1
from sync import Sync
2
3
sync = Sync()  # picks up SYNC_API_KEY from environment

1
sync = Sync(api_key="your-api-key")

1
from sync import Sync
2
from sync.common import Audio, Video, GenerationOptions
3
4
sync = Sync()
5
6
response = sync.generations.create(
7
    input=[
8
        Video(url="https://your-cdn.com/video.mp4"),
9
        Audio(url="https://your-cdn.com/audio.wav"),
10
    ],
11
    model="lipsync-2",
12
    options=GenerationOptions(sync_mode="cut_off"),
13
    output_file_name="my_output",
14
)

1
from sync import Sync
2
from sync.common import Video, TTS, GenerationOptions
3
4
sync = Sync()
5
6
response = sync.generations.create(
7
    input=[
8
        Video(url="https://your-cdn.com/video.mp4"),
9
        TTS(
10
            provider={
11
                "name": "elevenlabs",
12
                "voiceId": "EXAVITQu4vr4xnSDxMaL",
13
                "script": "Hello, this is a demo of the Sync lip sync API.",
14
                "stability": 0.5,
15
                "similarityBoost": 0.75,
16
            }
17
        ),
18
    ],
19
    model="lipsync-2",
20
    options=GenerationOptions(sync_mode="cut_off"),
21
)

1
import time
2
3
generation = sync.generations.get(job_id)
4
while generation.status not in ["COMPLETED", "FAILED", "REJECTED"]:
5
    time.sleep(10)
6
    generation = sync.generations.get(job_id)

1
response = sync.generations.create(
2
    input=[
3
        Video(url="https://your-cdn.com/video.mp4"),
4
        Audio(url="https://your-cdn.com/audio.wav"),
5
    ],
6
    model="lipsync-2",
7
    webhook_url="https://your-server.com/webhook",
8
)

1
# List all generations
2
all_generations = sync.generations.list()
3
4
# Filter by status
5
completed = sync.generations.list(status="COMPLETED")
6
pending = sync.generations.list(status="PENDING")
7
8
for gen in completed:
9
    print(f"{gen.id} - {gen.status} - {gen.output_url}")

1
generation = sync.generations.get("your-generation-id")
2
3
print(f"Status: {generation.status}")
4
print(f"Model: {generation.model}")
5
print(f"Created: {generation.created_at}")
6
print(f"Duration: {generation.output_duration}s")
7
8
if generation.output_url:
9
    print(f"Output: {generation.output_url}")

1
estimates = sync.generations.estimate_cost(
2
    input=[
3
        Video(url="https://your-cdn.com/video.mp4"),
4
        Audio(url="https://your-cdn.com/audio.wav"),
5
    ],
6
    model="lipsync-2",
7
)
8
9
for estimate in estimates:
10
    print(f"Estimated frames: {estimate.estimated_frame_count}")
11
    print(f"Estimated cost: ${estimate.estimated_generation_cost:.2f}")

1
from sync.common import Video, Audio
2
3
Video(url="https://your-cdn.com/video.mp4")
4
Audio(url="https://your-cdn.com/audio.wav")

1
from sync.common import Video, Audio
2
3
Video(asset_id="550e8400-e29b-41d4-a716-446655440000")
4
Audio(asset_id="660e8400-e29b-41d4-a716-446655440001")

1
with open("video.mp4", "rb") as video_file, open("audio.wav", "rb") as audio_file:
2
    response = sync.generations.create_with_files(
3
        video=video_file,
4
        audio=audio_file,
5
        model="lipsync-2",
6
    )

1
# List assets with pagination
2
assets = sync.assets.list(limit=10, sort_by="dateDesc")
3
for asset in assets.items:
4
    print(f"{asset.name} ({asset.type}) - {asset.duration_seconds}s")
5
6
# Get a specific asset
7
asset = sync.assets.get("550e8400-e29b-41d4-a716-446655440000")
8
print(f"URL: {asset.url}")

1
from sync import Sync
2
from sync.core.api_error import ApiError
3
4
sync = Sync()
5
6
try:
7
    response = sync.generations.create(
8
        input=[
9
            Video(url="https://your-cdn.com/video.mp4"),
10
            Audio(url="https://your-cdn.com/audio.wav"),
11
        ],
12
        model="lipsync-2",
13
    )
14
except ApiError as e:
15
    print(f"Status: {e.status_code}")
16
    print(f"Error: {e.body}")

1
import time
2
from sync.core.api_error import ApiError
3
4
def create_with_retry(sync, max_retries=3, **kwargs):
5
    for attempt in range(max_retries):
6
        try:
7
            return sync.generations.create(**kwargs)
8
        except ApiError as e:
9
            if e.status_code in (429, 500, 503) and attempt < max_retries - 1:
10
                wait = 2 ** attempt * 5  # 5s, 10s, 20s
11
                print(f"Retrying in {wait}s (attempt {attempt + 1}/{max_retries})")
12
                time.sleep(wait)
13
            else:
14
                raise

1
import asyncio
2
from sync import AsyncSync
3
from sync.common import Audio, Video
4
5
async def main():
6
    sync = AsyncSync()
7
8
    response = await sync.generations.create(
9
        input=[
10
            Video(url="https://your-cdn.com/video.mp4"),
11
            Audio(url="https://your-cdn.com/audio.wav"),
12
        ],
13
        model="lipsync-2",
14
    )
15
16
    job_id = response.id
17
    print(f"Job submitted: {job_id}")
18
19
    # Poll until complete
20
    generation = await sync.generations.get(job_id)
21
    while generation.status not in ["COMPLETED", "FAILED", "REJECTED"]:
22
        await asyncio.sleep(10)
23
        generation = await sync.generations.get(job_id)
24
25
    if generation.status == "COMPLETED":
26
        print(f"Output: {generation.output_url}")
27
28
asyncio.run(main())

1
from sync.common import (
2
    Audio,
3
    Video,
4
    TTS,
5
    GenerationOptions,
6
    Generation,
7
    GenerationStatus,
8
    Model,
9
)
10
11
# Response objects are typed -- your editor knows all available fields
12
generation: Generation = sync.generations.get("job-id")
13
status: GenerationStatus = generation.status  # "PENDING" | "PROCESSING" | "COMPLETED" | "FAILED" | "REJECTED"
14
model: Model = generation.model               # "lipsync-2" | "lipsync-2-pro" | ...
15
output_url: str | None = generation.output_url