Error Handling

1
import time
2
import requests
3
4
def api_request(method, url, headers, **kwargs):
5
    """Make an API request with exponential backoff retry."""
6
    max_retries = 3
7
    base_delay = 1
8
9
    for attempt in range(max_retries):
10
        response = requests.request(method, url, headers=headers, **kwargs)
11
12
        if response.status_code == 429:
13
            delay = base_delay * (2 ** attempt)
14
            print(f"Rate limited, retrying in {delay}s...")
15
            time.sleep(delay)
16
            continue
17
18
        if response.status_code >= 500:
19
            delay = base_delay * (2 ** attempt)
20
            print(f"Server error, retrying in {delay}s...")
21
            time.sleep(delay)
22
            continue
23
24
        return response
25
26
    return response  # Return last response after all retries

1
import time
2
3
def wait_for_ready(url, headers, interval=5, timeout=600):
4
    """Poll a resource until ready or failed."""
5
    elapsed = 0
6
    while elapsed < timeout:
7
        response = requests.get(url, headers=headers)
8
        resource = response.json()
9
        status = resource["status"]
10
11
        if status == "ready":
12
            return resource
13
        elif status == "failed":
14
            raise Exception(f"Resource failed: {resource}")
15
16
        print(f"  Status: {status} ({elapsed}s elapsed)")
17
        time.sleep(interval)
18
        elapsed += interval
19
20
    raise TimeoutError(f"Not ready after {timeout}s")

1
BASE_URL = "https://api.twelvelabs.io/v1.3"
2
3
# Wait for asset
4
asset = wait_for_ready(f"{BASE_URL}/assets/{asset_id}", HEADERS, interval=5)
5
6
# Wait for knowledge store item (longer interval - indexing takes more time)
7
item = wait_for_ready(
8
    f"{BASE_URL}/knowledge-stores/{store_id}/items/{item_id}",
9
    HEADERS,
10
    interval=10,
11
    timeout=600
12
)

1
import concurrent.futures
2
3
def wait_for_items(store_id, item_ids):
4
    """Wait for multiple knowledge store items in parallel."""
5
    def wait_one(item_id):
6
        return wait_for_ready(
7
            f"{BASE_URL}/knowledge-stores/{store_id}/items/{item_id}",
8
            HEADERS,
9
            interval=10
10
        )
11
12
    with concurrent.futures.ThreadPoolExecutor(max_workers=5) as executor:
13
        futures = {executor.submit(wait_one, iid): iid for iid in item_ids}
14
        results = {}
15
        for future in concurrent.futures.as_completed(futures):
16
            item_id = futures[future]
17
            results[item_id] = future.result()
18
19
    return results