11. Batch import ClipGT files from S3 using the Python SDK

Use the Python SDK to configure S3 access and trigger batch ClipGT imports programmatically.

11.1 Prerequisites and setup

The Foretify Manager service must have read access to the S3 bucket before triggering an import.

11.1.1 Required S3 permissions

The IAM principal used by the Foretify Manager service requires the following permissions on the bucket:

s3:ListBucket — to enumerate objects under a prefix
s3:GetObject — to read clip files during import

11.1.2 Credential configuration

Foretify Manager looks for S3 credentials in this order:

The Manager's application properties (aws.accessKeyId / aws.secretKey)
Environment variables: AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY
Named AWS profile (~/.aws/credentials)
The host's IAM role (via the instance metadata service)

Note

This credential configuration is used by Foretify Manager to validate the path and list files. The actual clip download runs in the import job pod via aws s3 cp, which uses the pod's own credentials. In most setups these are the same credentials.

11.2 Python SDK

Use launch_batch_import_clipgt_from_cloud_storage() from ftx.shell.flow_executions to trigger a batch ClipGT import from S3.

Provide exactly one of exact_paths or root_path:

Python: ClipGT import from S3 URI list

from argparse import ArgumentParser
from ftx._common.utils import add_common_arguments
import time
import sys


def parse():
    parser = ArgumentParser(description="Launch batch ClipGT import from an explicit S3 URI list")
    add_common_arguments(parser)
    parser.add_argument("--project_id", nargs="?", help="target project id")
    return parser.parse_args()


args = parse()
globals().update(vars(args))
project_id = args.project_id

from ftx.shell import client
client.login(host=host, user=user, port=port, https=https)

poll_interval = 5
poll_timeout_seconds = 600


from ftx.shell import flow_executions

# Import from a list of explicit S3 URIs
created = flow_executions.launch_batch_import_clipgt_from_cloud_storage(
    project_id=project_id,
    exact_paths=["s3://my-bucket/clips/clip1.zip", "s3://my-bucket/clips/clip2.zip"],
)

execution_id = created.get("id")
print(f"Flow execution id: {execution_id}")

terminal_statuses = {"COMPLETED", "FAILED"}
last_status = None
start_time = time.time()

while True:
    if time.time() - start_time > poll_timeout_seconds:
        print(
            f"Timed out after {poll_timeout_seconds}s waiting for execution "
            f"{execution_id} (last status: {last_status})"
        )
        sys.exit(5)

    details = flow_executions.get_by_id(execution_id)
    status = details.get("status")

    if status != last_status:
        print(f"Execution {execution_id} status: {status}")
        last_status = status

    if status in terminal_statuses:
        if status == "COMPLETED":
            print("Flow execution completed successfully")
            sys.exit(0)
        print("Flow execution failed")
        sys.exit(4)

    time.sleep(poll_interval)

Python: ClipGT import from S3 prefix

from argparse import ArgumentParser
from ftx._common.utils import add_common_arguments


def parse():
    parser = ArgumentParser(description="Launch batch ClipGT import from an S3 prefix")
    add_common_arguments(parser)
    parser.add_argument("--project_id", nargs="?", help="target project id")
    return parser.parse_args()


args = parse()
globals().update(vars(args))
project_id = args.project_id

from ftx.shell import client, flow_executions
client.login(host=host, user=user, port=port, https=https)



# Import all files under an S3 prefix
flow_executions.launch_batch_import_clipgt_from_cloud_storage(
    project_id=project_id,
    root_path="s3://my-bucket/clips/",
)

11.2.1 Parameters

Parameter	Type	Description
`project_id`	`str`	ID of the target project
`exact_paths`	`list[str]`	Explicit `s3://` URIs. Mutually exclusive with `root_path`
`root_path`	`str`	S3 prefix. Foretify Manager enumerates files under it. Mutually exclusive with `exact_paths`
`validate`	`bool`	Pre-flight validate `root_path` before creating assets. Default: `True`. No-op for `exact_paths`
`test_run_group_name`	`str`	Optional test run group name. Auto-generated if omitted
`run_name_prefix`	`str`	Optional prefix for individual run names
`labels`	`list[str]`	Optional labels to apply
`stationary_extrapolation`	`bool`	Enable stationary extrapolation. Default: `False`
`environment_settings_id`	`str`	ID of saved execution settings
`environment_settings`	`dict`	Inline execution settings. Mutually exclusive with `environment_settings_id`

Returns the created flow execution as a dict.

11.3 Error handling

11.3.1 Path validation errors

When validate=True (default) and root_path is provided, the SDK validates the path before creating any assets. If validation fails, a ValueError is raised with one of the following message codes:

Error	Cause
`CANNOT_ACCESS`	The bucket is not accessible. Check credentials and bucket permissions
`LIMIT_EXCEEDED`	The path contains more files than the allowed limit
`UNKNOWN_ERROR`	Validation failed for an unexpected reason

Python: ClipGT root_path validation error handling

from argparse import ArgumentParser
from ftx._common.utils import add_common_arguments


def parse():
    parser = ArgumentParser(description="Handle ClipGT root_path pre-flight validation errors")
    add_common_arguments(parser)
    parser.add_argument("--project_id", nargs="?", help="target project id")
    return parser.parse_args()


args = parse()
globals().update(vars(args))
project_id = args.project_id

from ftx.shell import client
client.login(host=host, user=user, port=port, https=https)


from ftx.shell import flow_executions

try:
    flow_executions.launch_batch_import_clipgt_from_cloud_storage(
        project_id=project_id,
        root_path="s3://my-bucket/clips/",
    )
except ValueError as e:
    print(e)
# Pre-flight validation failed for root_path 's3://my-bucket/clips/': CANNOT_ACCESS

11.3.2 Invalid path arguments

If both exact_paths and root_path are provided, or neither is provided, the SDK raises:

ValueError: Provide exactly one of exact_paths or root_path

11.3.3 Server-side errors

Server-side failures raise FmanagerServerException (from ftx.model.exceptions), which includes a status_code, message, and message_code.

Once your clips are imported, you can query the resulting runs using evaluation attributes — including clip_id and original_clipgt_url — through the Python SDK. See Evaluation attributes in SDK.