Error Handling & Best Practices

def is_prediction_error(obj: dict) -> bool:
    return "status" in obj and obj.get("status") == "failed" and "error_id" in obj

def is_inference_error(obj: dict) -> bool:
    return "error" in obj and isinstance(obj["error"], dict)

def handle_response(obj: dict):
    if is_inference_error(obj):
        err = obj["error"]
        raise InferenceAPIError(
            message=err["message"],
            code=err["code"],
            status=err["status"],
            request_id=err.get("request_id"),
        )
    if is_prediction_error(obj):
        raise PredictionAPIError(
            message=obj["error_message"],
            error_id=obj["error_id"],
            prediction_id=obj["id"],
        )

import time
import random
import httpx

def call_with_retry(fn, max_attempts: int = 4):
    """
    Retry a callable with exponential backoff + jitter.
    Retryable: 429, 500, 503
    Non-retryable: 400, 401, 402, 404, 422
    """
    NON_RETRYABLE = {400, 401, 402, 404, 422}
    
    for attempt in range(max_attempts):
        try:
            return fn()
        except httpx.HTTPStatusError as e:
            status = e.response.status_code
            if status in NON_RETRYABLE:
                raise  # Don't retry client errors
            
            if attempt == max_attempts - 1:
                raise  # Final attempt — surface the error
            
            # Respect Retry-After if present
            retry_after = e.response.headers.get("Retry-After")
            if retry_after:
                time.sleep(float(retry_after))
            else:
                # Exponential backoff: 1s, 2s, 4s, 8s … with ±25% jitter
                base_delay = 2 ** attempt
                jitter = random.uniform(0.75, 1.25)
                time.sleep(base_delay * jitter)

# Usage
result = call_with_retry(
    lambda: client.chat.completions.create(
        model="deepbrain-router",
        messages=[{"role": "user", "content": "Hello"}],
    )
)

import httpx, os

resp = httpx.post(
    "https://api.skytells.ai/v1/predictions",
    headers={"x-api-key": os.environ["SKYTELLS_API_KEY"]},
    json={
        "model_id": "stable-video-6s",
        "inputs": {"prompt": "A drone shot over a mountain lake at dawn"},
        "webhook_url": "https://yourapp.com/api/webhooks/skytells",
        "webhook_events": ["prediction.completed", "prediction.failed"],
    },
)
print(resp.json()["id"])  # prediction ID — store this

from fastapi import FastAPI, Request, Header, HTTPException
import hmac, hashlib

app = FastAPI()
WEBHOOK_SECRET = os.environ["SKYTELLS_WEBHOOK_SECRET"]  # set in Dashboard

@app.post("/api/webhooks/skytells")
async def handle_webhook(
    request: Request,
    x_skytells_signature: str = Header(None),
):
    body = await request.body()
    
    # Verify signature — never skip this in production
    expected = hmac.new(WEBHOOK_SECRET.encode(), body, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(f"sha256={expected}", x_skytells_signature or ""):
        raise HTTPException(status_code=401, detail="Invalid signature")
    
    event = await request.json()
    prediction = event["data"]
    
    if prediction["status"] == "succeeded":
        output_urls = prediction["output"]  # download within 24 hours
        await save_to_cloud_storage(output_urls)
    elif prediction["status"] == "failed":
        await log_prediction_failure(prediction["id"], prediction["error_id"])

import httpx, boto3, os
from urllib.parse import urlparse
from pathlib import Path

def save_outputs(prediction: dict, bucket: str) -> list[str]:
    """Download Prediction outputs to S3 and return permanent URLs."""
    s3 = boto3.client("s3")
    permanent_urls = []
    
    for i, url in enumerate(prediction.get("output", [])):
        # Determine file extension from URL
        ext = Path(urlparse(url).path).suffix or ".bin"
        key = f"predictions/{prediction['id']}/{i}{ext}"
        
        # Stream-download and upload
        with httpx.stream("GET", url) as r:
            s3.upload_fileobj(r.raw, bucket, key)
        
        permanent_urls.append(f"https://{bucket}.s3.amazonaws.com/{key}")
    
    return permanent_urls