Video Management

1
import requests
2
import json
3
4
url = f"{BASE_URL}/v1/videos/upload"
5
video_path = "/content/demo.mp4" # local video path
6
7
# Video indexing request body for a local video
8
data = {
9
   "index": True, # boolean value indicating whether the video should be indexed for search/qa/etc
10
   "enable_thumbnails": False, # boolean value indicating whether to generate thumbnails for video chunks, this will increase the indexing time but will allow for quick thumbnail loading at search time
11
   "video_name": "vid16", # name of video to store in vision agent system for informative purpose. not ID
12
   "video_start_absolute_timestamp": "2025-04-12T19:05:45", # (Optional) a absolute timestamp indicating video start time in ISO 8601 format
13
   "group_id": "20f4bc2d-3ebe-4fd2-829f-9d88c79e8a37", # optional video group assignment
14
}
15
headers = {
16
    "X-Api-Key": REKA_API_KEY
17
}
18
19
# Open the video file and send the request
20
with open(video_path, "rb") as file:
21
    files = {"file": (video_path[1:], file, "video/mp4")}  # Send as multipart/form-data
22
    response = requests.post(url, headers=headers, data=data, files=files)
23
24
# Print response
25
print(response.status_code, response.json())

1
import requests
2
import json
3
4
url = f"{BASE_URL}/v1/videos/upload"
5
video_url = "https://www.youtube.com/watch?v=dQw4w9WgXcQ" # video url
6
7
# Video indexing request body for a video url
8
data = {
9
   "index": True, 
10
   "video_name": "vid16", 
11
   "video_url": video_url,
12
   "video_start_absolute_timestamp": "2025-04-12T19:05:45",
13
   "group_id": "default",
14
}
15
headers = {
16
    "X-Api-Key": REKA_API_KEY
17
}
18
19
# Send the request
20
response = requests.post(url, headers=headers, data=data)
21
22
# Print response
23
print(response.status_code, response.json())

1
import os
2
import asyncio
3
import httpx
4
import time
5
import json
6
async def bulk_upload_videos(video_paths: list[str], api_key: str, caption_mode: str = "tagging_ad_video", max_concurrency: int = 2):
7
    tasks = []
8
    semaphore = asyncio.Semaphore(max_concurrency)
9
10
    async def limited_send_upload_request(sem: asyncio.Semaphore, video_path: str, *args, **kwargs):
11
        async with sem:
12
            response = await send_upload_request(video_path, *args, **kwargs)
13
            return video_path, response
14
15
    for video_path in video_paths:
16
        task = asyncio.create_task(limited_send_upload_request(semaphore, video_path, api_key, caption_mode))
17
        tasks.append(task)
18
19
    results = await asyncio.gather(*tasks)
20
21
    video_map = {}
22
    for video_path, result in results:
23
        if result and result.status_code == 200:
24
            try:
25
                video_id = result.json().get('video_id')
26
                if video_id:
27
                    video_map[video_path] = video_id
28
            except json.JSONDecodeError:
29
                print(f"Could not decode JSON from response for {video_path}")
30
    return video_map

1
import requests
2
3
video_id = "550e8400-e29b-41d4-a716-446655440000"
4
url = f"{BASE_URL}/v1/videos/{video_id}"
5
headers = {
6
    "X-Api-Key": REKA_API_KEY
7
}
8
9
response = requests.get(url, headers=headers)
10
print(response.status_code, response.json())

1
import requests
2
3
url = f"{BASE_URL}/v1/videos"
4
headers = {
5
    "X-Api-Key": REKA_API_KEY
6
}
7
8
response = requests.get(url, headers=headers)
9
print(response.status_code, response.json())

1
import requests
2
3
video_id = "550e8400-e29b-41d4-a716-446655440000"
4
url = f"{BASE_URL}/v1/videos/{video_id}"
5
headers = {
6
    "X-Api-Key": REKA_API_KEY
7
}
8
9
response = requests.delete(url, headers=headers)
10
print(response.status_code, response.json())