Video Management

1
import requests
2
import json
3
4
url = f"{BASE_URL}/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
}
14
headers = {
15
    "X-Api-Key": REKA_API_KEY
16
}
17
18
# Open the video file and send the request
19
with open(video_path, "rb") as file:
20
    files = {"file": (video_path[1:], file, "video/mp4")}  # Send as multipart/form-data
21
    response = requests.post(url, headers=headers, data=data, files=files)
22
23
# Print response
24
print(response.status_code, response.json())

1
import requests
2
import json
3
4
url = f"{BASE_URL}/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
}
14
headers = {
15
    "X-Api-Key": REKA_API_KEY
16
}
17
18
# Send the request
19
response = requests.post(url, headers=headers, data=data)
20
21
# Print response
22
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
import json
3
4
url = f"{BASE_URL}/videos/get"
5
6
# Video get request body
7
data = {
8
   "video_ids": ["550e8400-e29b-41d4-a716-446655440000"]
9
}
10
headers = {
11
    "X-Api-Key": REKA_API_KEY
12
}
13
14
# Send the request
15
response = requests.post(url, headers=headers, data=data)
16
17
# Print response
18
print(response.status_code, response.json())

1
import requests
2
import json
3
4
url = f"{BASE_URL}/videos/get"
5
6
headers = {
7
    "X-Api-Key": REKA_API_KEY
8
}
9
10
# Send the request
11
response = requests.post(url, headers=headers)
12
13
# Print response
14
print(response.status_code, response.json())

1
import requests
2
import json
3
4
url = f"{BASE_URL}/videos/delete"
5
6
# Video delete request body
7
data = {
8
   "video_ids": ["550e8400-e29b-41d4-a716-446655440000"] # List of video ids to delete
9
}
10
headers = {
11
    "X-Api-Key": REKA_API_KEY
12
}
13
14
# Send the request
15
response = requests.delete(url, headers=headers, data=data)
16
17
# Print response
18
print(response.status_code, response.json())