How to Use OpenAI's Sora API: A Comprehensive Guide

class SoraClient:
    def __init__(self, api_key=None):
        self.api_key = api_key or os.getenv("OPENAI_API_KEY")
        self.api_version = self._detect_api_version()
        
    def _detect_api_version(self):
        """Detect the current Sora API version"""
        try:
            # Make a minimal API call to check version
            metadata = openai.Sora.get_info()
            return metadata.version
        except:
            # Fall back to default version if detection fails
            return "v1"
    
    def generate_video(self, prompt, duration, **kwargs):
        """Version-aware video generation"""
        if self._supports_feature("high_resolution") and kwargs.get("high_res"):
            resolution = "1080p"
        else:
            resolution = "720p"
            
        if self._supports_feature("extended_duration") and duration > 60:
            # Handle with segmentation for older API versions
            return self._generate_segmented(prompt, duration, **kwargs)
        
        # Standard generation with version-appropriate parameters
        params = self._prepare_parameters(prompt, duration, **kwargs)
        return openai.Sora.create(**params)
        
    def _supports_feature(self, feature_name):
        """Check if current API version supports a specific feature"""
        feature_map = {
            "high_resolution": ["v1.2", "v2.0"],
            "extended_duration": ["v2.0"],
            "style_transfer": ["v1.5", "v2.0"]
        }
        
        if feature_name in feature_map:
            return self.api_version in feature_map[feature_name]
        return False
        
    def _prepare_parameters(self, prompt, duration, **kwargs):
        """Prepare version-appropriate parameters"""
        # Base parameters supported across versions
        params = {
            "prompt": prompt,
            "duration_seconds": min(duration, 60)  # Enforce limits for older versions
        }
        
        # Add version-specific parameters
        if self.api_version >= "v1.5" and "style" in kwargs:
            params["style_preset"] = kwargs["style"]
            
        # Add other parameters based on version capability
        
        return params

import asyncio
import aiohttp
import time
from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
app = FastAPI()
class VideoRequest(BaseModel):
    prompt: str
    duration: int
    callback_url: str
    user_id: str
# Simple in-memory queue for demonstration
request_queue = asyncio.Queue()
processing_semaphore = asyncio.Semaphore(5)  # Limit concurrent processing
@app.post("/generate")
async def enqueue_generation(request: VideoRequest, background_tasks: BackgroundTasks):
    # Add to queue
    await request_queue.put(request)
    
    # Start processing in background if not already running
    background_tasks.add_task(process_queue)
    
    return {"status": "queued", "queue_position": request_queue.qsize()}
async def process_queue():
    while not request_queue.empty():
        async with processing_semaphore:
            request = await request_queue.get()
            
            try:
                # Generate video
                response = await generate_video_async(request.prompt, request.duration)
                
                # Notify via callback
                await send_callback(request.callback_url, {
                    "user_id": request.user_id,
                    "status": "completed",
                    "video_url": response.data[0].url
                })
                
            except Exception as e:
                # Handle failures
                await send_callback(request.callback_url, {
                    "user_id": request.user_id,
                    "status": "failed",
                    "error": str(e)
                })
            
            finally:
                request_queue.task_done()
                
async def generate_video_async(prompt, duration):
    """Asynchronous video generation"""
    # In a real implementation, use the OpenAI async client
    return openai.Sora.create(
        prompt=prompt,
        duration_seconds=duration
    )
async def send_callback(url, data):
    """Send callback to notify of completion"""
    async with aiohttp.ClientSession() as session:
        await session.post(url, json=data)