1
from twelvelabs import TwelveLabs
2
from twelvelabs.indexes import IndexesCreateRequestModelsItem
3
from twelvelabs.tasks import TasksRetrieveResponse
4
5
# 1. Initialize the client
6
client = TwelveLabs(api_key="<YOUR_API_KEY>")
7
8
# 2. Create an index
9
# An index is a container for organizing your video content
10
index = client.indexes.create(
11
    index_name="<YOUR_INDEX_NAME>",
12
    models=[
13
        IndexesCreateRequestModelsItem(
14
            model_name="marengo3.0",
15
            model_options=["visual", "audio"]
16
        )
17
    ]
18
)
19
if index.id is None:
20
    raise RuntimeError("Failed to create an index.")
21
print(f"Created index: id={index.id}")
22
23
# 3. Upload a video
24
task = client.tasks.create(
25
    index_id=index.id,
26
    video_url="<YOUR_VIDEO_URL>"
27
    # Or for a local file: video_file=open("<PATH_TO_VIDEO_FILE>", "rb")
28
    )
29
print(f"Created task: id={task.id}")
30
31
# 4. Monitor the indexing process
32
def on_task_update(task: TasksRetrieveResponse):
33
    print(f"  Status={task.status}")
34
task = client.tasks.wait_for_done(task_id=task.id, callback=on_task_update)
35
if task.status != "ready":
36
    raise RuntimeError(f"Indexing failed with status {task.status}")
37
print(
38
    f"Upload complete. The unique identifier of your video is {task.video_id}.")
39
40
# 5. Perform a search request
41
search_pager = client.search.query(
42
    index_id=index.id, query_text="<YOUR_QUERY>", search_options=["visual", "audio"],)
43
44
# 6. Process the search results
45
print("Search results:")
46
for clip in search_pager:
47
    print(
48
        f" video_id {clip.video_id} rank={clip.rank} start={clip.start} end={clip.end}"
49
    )