> For clean Markdown of any page, append .md to the page URL.
> For a complete documentation index, see https://docs.twelvelabs.io/llms.txt.
> For AI client integration (Claude Code, Cursor, etc.), connect to the MCP server at https://docs.twelvelabs.io/_mcp/server.

# Backblaze B2 - Media management application

> Add video search to media asset management using TwelveLabs and Backblaze B2.

**Summary**: The [Media Asset Management example application](https://github.com/backblaze-b2-samples/b2-twelvelabs-example) demonstrates how you can add video understanding capabilities to a typical media asset management application using [TwelveLabs](https://www.twelvelabs.io) for video understanding and [Backblaze B2](https://www.backblaze.com/cloud-storage) for cloud storage.

**Description**: The application allows you to upload videos and perform deep semantic searches across multiple modalities such as visual, conversation, text-in-video, and logo. The matching video segments in the response are grouped by video, and you can view details about each segment.

**Code explanation**: The  [b2-twelvelabs-example](https://github.com/backblaze-b2-samples/b2-twelvelabs-example) GitHub repository provides detailed instructions on setting up the example application on your computer and using it.

**GitHub**: [Backblaze B2 + TwelveLabs Media Asset Management Example](https://github.com/backblaze-b2-samples/b2-twelvelabs-example)

![](/_fern-img/30c65e44ee0437b6f990144746d6eb0eebefd0402d9edb4a633c8752fb8b7b88.webp)

# Integration with TwelveLabs

The integration with TwelveLabs video enhances the functionalities of a typical media management application by adding video understanding capabilities. The process is segmented into two main steps:

* Upload and index videos
* Search videos

# Upload and index videos

A [Huey](https://huey.readthedocs.io/en/latest/) task automates the process of uploading and indexing videos. For each video, the application invokes the `create` method of the `task` object with the following parameters and values:

* `index_id`: A string representing unique identifier of the index to which the video will be updated.
* `url`: A string representing the URL of the video to be uploaded.
* `disable_video_stream`: A boolean indicating that the platform shouldn't store the video for streaming.

**`Python`**

```python Python
@huey.db_task()
def do_video_indexing(video_tasks):
    print(f'Creating tasks: {video_tasks}')

    # Create a task for each video we want to index
    for video_task in video_tasks:
        task = TWELVE_LABS_CLIENT.task.create(
            TWELVE_LABS_INDEX_ID,
            url=default_storage.url(video_task['video']),
            disable_video_stream=True
        )
        print(f'Created task: {task}')
        video_task['task_id'] = task.id

    print(f'Created {len(video_tasks)} tasks')

```

Then, the application monitors the status of the upload process by invoking the `retrieve` method of the `task` object with the unique identifier of a task as a parameter:

**`Python`**

```python Python
    print(f'Polling TwelveLabs for {video_tasks}')

    # Do a single database query for all the videos we're interested in
    video_ids = [video_task['id'] for video_task in video_tasks]
    videos = Video.objects.filter(id__in=video_ids)

    while True:
        done = True
        videos_to_save = []

        # Retrieve status for each task we created
        for video_task in video_tasks:
            # What's our current state for this video?
            video = videos.get(video__exact=video_task['video'])

            # Do we still need to retrieve status for this task?
            if video.status != 'Ready':
                task = TWELVE_LABS_CLIENT.task.retrieve(video_task['task_id'])
                if task.status != 'ready':
                    # We'll need to go round the loop again
                    done = False

                # Do we need to write a new status to the DB?
                if video.status.lower() != task.status:
                    # We store the status in the DB in title case, so it's ready to render on the page
                    new_status = task.status.title()
                    print(f'Updating status for {video_task["video"]} from {video.status} to {new_status}')
                    video.status = new_status
                    if task.status == 'ready':
                        video.video_id = task.video_id
                        get_all_video_data(video)
                    videos_to_save.append(video)

        if len(videos_to_save) > 0:
            Video.objects.bulk_update(videos_to_save, ['status', 'video_id', THUMBNAILS_PATH, TRANSCRIPTS_PATH, TEXT_PATH, LOGOS_PATH])

        if done:
            break

        sleep(TWELVE_LABS_POLL_INTERVAL)

    print(f'Done polling {video_tasks}')
```

## Search videos

![](/_fern-img/951fbb79ae5a561d8f476d97d7f3812ff6fb44d43b72df5d8f92072923acb11e.webp)

\


The application invokes the `query` method of the `search` object with the following parameters:

* `index_id`: A string representing the unique identifier of the index containing the videos to be searched.
* `query`: A string representing the query the user has provided.
* `options`: An array of strings representing the sources of information the TwelveLabs video understanding platform should consider when performing the search.
* `group_by`: A string specifying that the matching video clips in the response must be grouped by video.
* `threshold`: A string specifying the sstrictness of the thresholds for assigning the high, medium, or low confidence levels to search results. See the [Filter on the level of confidence](/docs/guides/search/filtering#filtering-on-the-level-of-confidence) section for details.

**`Python`**

```python Python
def get_queryset(self):
    """
    Search TwelveLabs for videos matching the query
    """
    query = self.request.GET.get("query", None)

    result = TWELVE_LABS_CLIENT.search.query(
        TWELVE_LABS_INDEX_ID,
        query,
        ["visual", "conversation", "text_in_video", "logo"],
        group_by="video",
        threshold="medium"
    )

    # Search results may be in multiple pages, so we need to loop until we're done retrieving them
    search_data = result.data
    print(f"First page's data: {search_data}")

    search_results = []
    while True:
        # Do a database query to get the videos for each page of results
        video_ids = [group.id for group in search_data]
        videos = Video.objects.filter(video_id__in=video_ids)
        for group in search_data:
            try:
                search_results.append(SearchResult(video=videos.get(video_id__exact=group.id),
                                                   clip_count=len(group.clips),
                                                   clips=group.clips.model_dump_json()))
            except self.model.DoesNotExist:
                # There is a video in TwelveLabs, but no corresponding row in the database.
                # Just report it and carry on.
                print(f'Can\'t find match for video_id {group.id}')

        # Is there another page?
        try:
            search_data = next(result)
            print(f"Next page's data: {search_data}")
        except StopIteration:
            print("There is no next page in search result")
            break

    return search_results
```

# Next steps

After reading this page, you have several options:

* **Customize and use the example application**: Explore the [Media Asset Management example application](https://github.com/backblaze-b2-samples/b2-twelvelabs-example) on GitHub to understand its features and implementation. You can make changes to the application and add more functionalities to suit your specific use case.
* **Explore further**: Try the [applications built by the community](/docs/resources/from-the-community) or our [sample applications](/docs/resources/sample-applications) to get more insights into the TwelveLabs Video Understanding Platform's diverse capabilities and learn more about integrating the platform into your applications.