Video-to-GIF conversion in Python using Terraform

Written by

Yandex Cloud

Updated at April 6, 2026

Get your cloud ready
- Required paid resources
Create an infrastructure
Create a table
Test the application
How to delete the resources you created

To create a video-to-GIF conversion framework in Python using Terraform:

Get your cloud ready

Navigate to the management console and log in to Yandex Cloud or create a new account.
On the Yandex Cloud Billing page, make sure you have a billing account linked and it has the ACTIVE or TRIAL_ACTIVE status. If you do not have a billing account, create one and link a cloud to it.

If you have an active billing account, you can create or select a folder for your infrastructure on the cloud page.

Learn more about clouds and folders here.

Required paid resources

The infrastructure support cost includes:

Fee for invoking functions (see Yandex Cloud Functions pricing).
Fee for running queries to the database (see Yandex Managed Service for YDB pricing).
Fee for storing data in a bucket (see Yandex Object Storage pricing).

With Terraform, you can quickly create a cloud infrastructure in Yandex Cloud and manage it using configuration files. These files store the infrastructure description written in HashiCorp Configuration Language (HCL). If you change the configuration files, Terraform automatically detects which part of your configuration is already deployed, and what should be added or removed.

Terraform is distributed under the Business Source License. The Yandex Cloud provider for Terraform is distributed under the MPL-2.0 license.

For more information about the provider resources, see the relevant documentation on the Terraform website or its mirror.

To create an infrastructure using Terraform:

Install Terraform, get the credentials, and specify the source for installing Yandex Cloud (see Configure your provider, step 1).

Prepare your infrastructure description files:

Ready-made configuration

Manually

Clone the repository with configuration files.

git clone https://github.com/yandex-cloud-examples/yc-serverless-video-gif-converter.git

Navigate to the repository directory. It should now contain the following files:
- video-converting.tf: New infrastructure configuration.
- ffmpeg-api.zip: API function archive.
- src.zip: Converter function archive.

Create a folder for configuration files.

In the folder, create:

video-converting.tf configuration file:

video-converting.tf

# Declaring variables

locals {
  folder_id   = "<folder_ID>"
  bucket_name = "<bucket_name>"
}

# Configuring the provider

terraform {
  required_providers {
    yandex = {
      source  = "yandex-cloud/yandex"
    }
  }
}

provider "yandex" {
  folder_id = local.folder_id
}

# Creating a service account and assigning roles to it

resource "yandex_iam_service_account" "sa" {
  name = "ffmpeg-sa"
}

resource "yandex_resourcemanager_folder_iam_member" "ymq-reader" {
  folder_id = local.folder_id
  role      = "ymq.reader"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

resource "yandex_resourcemanager_folder_iam_member" "ymq-writer" {
  folder_id = local.folder_id
  role      = "ymq.writer"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

resource "yandex_resourcemanager_folder_iam_member" "lockbox-payload-viewer" {
  folder_id = local.folder_id
  role      = "lockbox.payloadViewer"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

resource "yandex_resourcemanager_folder_iam_member" "storage-editor" {
  folder_id = local.folder_id
  role      = "storage.editor"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

resource "yandex_resourcemanager_folder_iam_member" "storage-uploader" {
  folder_id = local.folder_id
  role      = "storage.uploader"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

resource "yandex_resourcemanager_folder_iam_member" "ydb-admin" {
  folder_id = local.folder_id
  role      = "ydb.admin"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

resource "yandex_resourcemanager_folder_iam_member" "serverless-functions-invoker" {
  folder_id = local.folder_id
  role      = "serverless.functions.invoker"
  member    = "serviceAccount:${yandex_iam_service_account.sa.id}"
}

# Creating a static key for the service account

resource "yandex_iam_service_account_static_access_key" "sa-static-key" {
  service_account_id = yandex_iam_service_account.sa.id
  description        = "static access key for database, message queue and bucket"
}

# Creating a secret

resource "yandex_lockbox_secret" "secretmq" {
  name = "ffmpeg-sa-secret"
}

resource "yandex_lockbox_secret_version" "my_version" {
  secret_id = yandex_lockbox_secret.secretmq.id
  entries {
    key        = "ACCESS_KEY_ID"
    text_value = yandex_iam_service_account_static_access_key.sa-static-key.access_key
  }
    entries {
    key        = "SECRET_ACCESS_KEY"
    text_value = yandex_iam_service_account_static_access_key.sa-static-key.secret_key
  }
}

# Creating a message queue

resource "yandex_message_queue" "converter_queue" {
  name                       = "converter-queue"
  visibility_timeout_seconds = 600
  message_retention_seconds  = 1209600
  receive_wait_time_seconds  = 20

  access_key = yandex_iam_service_account_static_access_key.sa-static-key.access_key
  secret_key = yandex_iam_service_account_static_access_key.sa-static-key.secret_key
  depends_on = [ yandex_resourcemanager_folder_iam_member.ymq-writer ]
}

# Creating a database

resource "yandex_ydb_database_serverless" "api_db" {
  name        = "db-converter"
  location_id = "ru-central1"
}

# Creating a bucket and uploading an archive

resource "yandex_storage_bucket" "conv_func_bucket" {
  folder_id  = local.folder_id
  access_key = yandex_iam_service_account_static_access_key.sa-static-key.access_key
  secret_key = yandex_iam_service_account_static_access_key.sa-static-key.secret_key
  bucket     = local.bucket_name
}

resource "yandex_storage_object" "archive" {
  access_key   = yandex_iam_service_account_static_access_key.sa-static-key.access_key
  secret_key   = yandex_iam_service_account_static_access_key.sa-static-key.secret_key
  bucket       = yandex_storage_bucket.conv_func_bucket.id
  key          = "src.zip"
  source       = "src.zip"
  content_type = "application/zip"
}

# Creating an API function

resource "yandex_function" "api-function" {
  name               = "ffmpeg-api"
  runtime            = "python312"
  user_hash          = filesha256("ffmpeg-api.zip")
  memory             = "256"
  entrypoint         = "index.handle_api"
  execution_timeout  = "5"
  service_account_id = yandex_iam_service_account.sa.id
  environment = {
    DOCAPI_ENDPOINT = yandex_ydb_database_serverless.api_db.document_api_endpoint
    YMQ_QUEUE_URL   = yandex_message_queue.converter_queue.id
    SECRET_ID       = yandex_lockbox_secret.secretmq.id
  }
  content {
    zip_filename = "ffmpeg-api.zip"
  }
}

# Creating a converter function

resource "yandex_function" "converter" {
  name               = "ffmpeg-converter"
  runtime            = "python312"
  user_hash          = filesha256("src.zip")
  memory             = "2048"
  entrypoint         = "index.handle_process_event"
  execution_timeout  = "600"
  service_account_id = yandex_iam_service_account.sa.id
  environment = {
    DOCAPI_ENDPOINT = yandex_ydb_database_serverless.api_db.document_api_endpoint
    YMQ_QUEUE_URL   = yandex_message_queue.converter_queue.id
    S3_BUCKET       = yandex_storage_bucket.conv_func_bucket.id
    SECRET_ID       = yandex_lockbox_secret.secretmq.id
  }
  package {
    bucket_name = yandex_storage_bucket.conv_func_bucket.id
    object_name = "src.zip"
  }
}

# Creating a trigger

resource "yandex_function_trigger" "converter_trigger" {
  name = "ffmpeg-trigger"
  message_queue {
    queue_id           = yandex_message_queue.converter_queue.arn
    service_account_id = yandex_iam_service_account.sa.id
    batch_size         = "1"
    batch_cutoff       = "10"
    visibility_timeout = 600
  }
  function {
    id                 = yandex_function.converter.id
    tag                = "$latest"
    service_account_id = yandex_iam_service_account.sa.id
  }
}

For an API function:

Create a file named index.py and paste this content into it:

index.py for an API function

import json
import os
import subprocess
import uuid
from urllib.parse import urlencode

import boto3
import requests
import yandexcloud
from yandex.cloud.lockbox.v1.payload_service_pb2 import GetPayloadRequest
from yandex.cloud.lockbox.v1.payload_service_pb2_grpc import PayloadServiceStub

boto_session = None
storage_client = None
docapi_table = None
ymq_queue = None


def get_boto_session():
    global boto_session
    if boto_session is not None:
        return boto_session

    # initialize lockbox and read secret value
    yc_sdk = yandexcloud.SDK()
    channel = yc_sdk._channels.channel("lockbox-payload")
    lockbox = PayloadServiceStub(channel)
    response = lockbox.Get(GetPayloadRequest(secret_id=os.environ['SECRET_ID']))

    # extract values from secret
    access_key = None
    secret_key = None
    for entry in response.entries:
        if entry.key == 'ACCESS_KEY_ID':
            access_key = entry.text_value
        elif entry.key == 'SECRET_ACCESS_KEY':
            secret_key = entry.text_value
    if access_key is None or secret_key is None:
        raise Exception("secrets required")
    print("Key id: " + access_key)

    # initialize boto session
    boto_session = boto3.session.Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key
    )
    return boto_session


def get_ymq_queue():
    global ymq_queue
    if ymq_queue is not None:
        return ymq_queue

    ymq_queue = get_boto_session().resource(
        service_name='sqs',
        endpoint_url='https://message-queue.api.cloud.yandex.net',
        region_name='ru-central1'
    ).Queue(os.environ['YMQ_QUEUE_URL'])
    return ymq_queue


def get_docapi_table():
    global docapi_table
    if docapi_table is not None:
        return docapi_table

    docapi_table = get_boto_session().resource(
        'dynamodb',
        endpoint_url=os.environ['DOCAPI_ENDPOINT'],
        region_name='ru-central1'
    ).Table('tasks')
    return docapi_table


def get_storage_client():
    global storage_client
    if storage_client is not None:
        return storage_client

    storage_client = get_boto_session().client(
        service_name='s3',
        endpoint_url='https://storage.yandexcloud.net',
        region_name='ru-central1'
    )
    return storage_client

# API handler

def create_task(src_url):
    task_id = str(uuid.uuid4())
    get_docapi_table().put_item(Item={
        'task_id': task_id,
        'ready': False
    })
    get_ymq_queue().send_message(MessageBody=json.dumps({'task_id': task_id, "src": src_url}))
    return {
        'task_id': task_id
    }


def get_task_status(task_id):
    task = get_docapi_table().get_item(Key={
        "task_id": task_id
    })
    if task['Item']['ready']:
        return {
            'ready': True,
            'gif_url': task['Item']['gif_url']
        }
    return {'ready': False}


def handle_api(event, context):
    action = event['action']
    if action == 'convert':
        return create_task(event['src_url'])
    elif action == 'get_task_status':
        return get_task_status(event['task_id'])
    else:
        return {"error": "unknown action: " + action}

Create a file named requirements.txt and specify these libraries in it:
```
boto3
yandexcloud
```
In the folder, create an archive named ffmpeg-api.zip containing the files requirements.txt and index.py.

For a converter function:

Create a file named index.py and paste this content into it:

index.py for a converter function

import json
import os
import subprocess
import uuid
from urllib.parse import urlencode

import boto3
import requests
import yandexcloud
from yandex.cloud.lockbox.v1.payload_service_pb2 import GetPayloadRequest
from yandex.cloud.lockbox.v1.payload_service_pb2_grpc import PayloadServiceStub

boto_session = None
storage_client = None
docapi_table = None
ymq_queue = None


def get_boto_session():
    global boto_session
    if boto_session is not None:
        return boto_session

    # initialize lockbox and read secret value
    yc_sdk = yandexcloud.SDK()
    channel = yc_sdk._channels.channel("lockbox-payload")
    lockbox = PayloadServiceStub(channel)
    response = lockbox.Get(GetPayloadRequest(secret_id=os.environ['SECRET_ID']))

    # extract values from secret
    access_key = None
    secret_key = None
    for entry in response.entries:
        if entry.key == 'ACCESS_KEY_ID':
            access_key = entry.text_value
        elif entry.key == 'SECRET_ACCESS_KEY':
            secret_key = entry.text_value
    if access_key is None or secret_key is None:
        raise Exception("secrets required")
    print("Key id: " + access_key)

    # initialize boto session
    boto_session = boto3.session.Session(
        aws_access_key_id=access_key,
        aws_secret_access_key=secret_key
    )
    return boto_session


def get_ymq_queue():
    global ymq_queue
    if ymq_queue is not None:
        return ymq_queue

    ymq_queue = get_boto_session().resource(
        service_name='sqs',
        endpoint_url='https://message-queue.api.cloud.yandex.net',
        region_name='ru-central1'
    ).Queue(os.environ['YMQ_QUEUE_URL'])
    return ymq_queue


def get_docapi_table():
    global docapi_table
    if docapi_table is not None:
        return docapi_table

    docapi_table = get_boto_session().resource(
        'dynamodb',
        endpoint_url=os.environ['DOCAPI_ENDPOINT'],
        region_name='ru-central1'
    ).Table('tasks')
    return docapi_table


def get_storage_client():
    global storage_client
    if storage_client is not None:
        return storage_client

    storage_client = get_boto_session().client(
        service_name='s3',
        endpoint_url='https://storage.yandexcloud.net',
        region_name='ru-central1'
    )
    return storage_client

# Converter handler

def download_from_ya_disk(public_key, dst):
    api_call_url = 'https://cloud-api.yandex.net/v1/disk/public/resources/download?' + \
                   urlencode(dict(public_key=public_key))
    response = requests.get(api_call_url)
    download_url = response.json()['href']
    download_response = requests.get(download_url)
    with open(dst, 'wb') as video_file:
        video_file.write(download_response.content)


def upload_and_presign(file_path, object_name):
    client = get_storage_client()
    bucket = os.environ['S3_BUCKET']
    client.upload_file(file_path, bucket, object_name)
    return client.generate_presigned_url('get_object', Params={'Bucket': bucket, 'Key': object_name}, ExpiresIn=3600)


def handle_process_event(event, context):
    for message in event['messages']:
        task_json = json.loads(message['details']['message']['body'])
        task_id = task_json['task_id']
        # Download video
        download_from_ya_disk(task_json['src'], '/tmp/video.mp4')
        # Convert with ffmpeg
        subprocess.run(['ffmpeg', '-i', '/tmp/video.mp4', '-r', '10', '-s', '320x240', '/tmp/result.gif'])
        result_object = task_id + ".gif"
        # Upload to Object Storage and generate presigned url
        result_download_url = upload_and_presign('/tmp/result.gif', result_object)
        # Update task status in DocAPI
        get_docapi_table().update_item(
            Key={'task_id': task_id},
            AttributeUpdates={
                'ready': {'Value': True, 'Action': 'PUT'},
                'gif_url': {'Value': result_download_url, 'Action': 'PUT'},
            }
        )
    return "OK"

Create a file named requirements.txt and specify these libraries in it:
```
boto3
requests
yandexcloud
```
Prepare the FFmpeg executable. On FFmpeg's official website, navigate to the Linux Static Builds section, download the 64-bit FFmpeg archive, and make the file executable by running the chmod +x ffmpeg command.
In the folder, create an archive named src.zip containing the files requirements.txt and index.py, and the FFmpeg executable.

Learn more about the properties of Terraform resources in the relevant provider guides:

In the video-converting.tf file, set the following user-defined properties:
- folder_id: Folder ID.
- bucket: Bucket name.
Create the resources:
1. In the terminal, go to the directory where you edited the configuration file.
2. Make sure the configuration file is correct using this command:
```
terraform validate
```
  If the configuration is correct, you will get this message:
```
Success! The configuration is valid.
```
3. Run this command:
```
terraform plan
```
  You will see a detailed list of resources. No changes will be made at this step. If the configuration contains any errors, Terraform will show them.
4. Apply the changes:
```
terraform apply
```
5. Type yes and press Enter to confirm the changes.

After you have created the infrastructure, create a table in YDB.

Create a table

Create a table in YDB:
- Name: tasks.
- Table type: Document table.
- Columns: One column named task_id of the String type. Set the Partition key attribute.

After you have created the table, test the application.

Test the application

Create a task

Management console

In the management console, select the folder containing the ffmpeg-api function.
Select Cloud Functions.
Select the ffmpeg-api function.
Navigate to the Testing tab.
In the Payload field, enter:
```
{"action":"convert", "src_url":"<link_to_video>"}
```
Where <link_to_video> is a link to an MP4 video file saved to Yandex Disk.
Click Run test.

You will see the task ID in the Function output field:

{ "task_id": "c4269ceb-8d3a-40fe-95f0-84cf********" }

View the queue statistics

Once the task is created, the queued message count will increase by one and a trigger will fire. Make sure messages arrive in the queue and are handled. To do this, view the queue statistics.

Management console

In the management console, select the folder housing converter-queue.
Select Message Queue.
Select converter-queue.
Under General information, you can see the queued and processed message counts.
Go to Monitoring. View the Overall queue stats charts.

View the function logs

The trigger should invoke the converter function for each message in the queue. To make sure the function is invoked, check its logs.

Management console

In the management console, select the folder containing the ffmpeg-converter function.
Select Cloud Functions.
Select ffmpeg-converter.
Go to the Logs tab and specify the period to view the logs for.

Get a link to a GIF file

Management console

In the management console, select the folder containing the ffmpeg-api function.
Select Cloud Functions.
Select the ffmpeg-api function.
Navigate to the Testing tab.

In the Payload field, enter the following request:

{"action":"get_task_status", "task_id":"<job_ID>"}

Click Run test.

If video conversion to GIF has not been completed, the Function output field will return:

{
    "ready": false
}

Otherwise, you will get a link to the GIF file:

{
    "ready": true,
    "gif_url": "https://storage.yandexcloud.net/<bucket_name>/1b4db1a6-f2b2-4b1c-b662-37f7********.gif?X-Amz-Algorithm=AWS4-HMAC-SHA256&X-Amz-Credential=qxLftbbZ91U695ysemyZ%2F202********ru-central1%2Fs3%2Faws4_request&X-Amz-Date=20210831T110351Z&X-Amz-Expires=3600&X-Amz-SignedHeaders=host&X-Amz-Signature=f4a5fe7848274a09be5b221fbf8a9f6f2b385708cfa351861a4e69df********"
}

How to delete the resources you created

To stop paying for the resources you created:

Open the video-converting.tf file and delete your infrastructure description from it.
Apply the changes:
1. In the terminal, go to the directory where you edited the configuration file.
2. Make sure the configuration file is correct using this command:
```
terraform validate
```
  If the configuration is correct, you will get this message:
```
Success! The configuration is valid.
```
3. Run this command:
```
terraform plan
```
  You will see a detailed list of resources. No changes will be made at this step. If the configuration contains any errors, Terraform will show them.
4. Apply the changes:
```
terraform apply
```
5. Type yes and press Enter to confirm the changes.

Video-to-GIF conversion in Python using Terraform

Get your cloud ready

Required paid resources

Create an infrastructure

Create a table

Test the application

Create a task

View the queue statistics

View the function logs

Get a link to a GIF file

How to delete the resources you created

See also

Was the article helpful?

Video-to-GIF conversion in Python using Terraform

Get your cloud readyGet your cloud ready

Required paid resourcesRequired paid resources

Create an infrastructureCreate an infrastructure

Create a tableCreate a table

Test the applicationTest the application

Create a taskCreate a task

View the queue statisticsView the queue statistics

View the function logsView the function logs

Get a link to a GIF fileGet a link to a GIF file

How to delete the resources you createdHow to delete the resources you created

See alsoSee also

Was the article helpful?

Get your cloud ready

Required paid resources

Create an infrastructure

Create a table

Test the application

Create a task

View the queue statistics

View the function logs

Get a link to a GIF file

How to delete the resources you created

See also