Frame.io Python SDK — 上传指南

本指南介绍如何使用 Frame.io Python SDK (frameio) 将文件上传到 Frame.io。 SDK 支持通过预签名 URL 向 S3 上传分片的多分段,具备并行处理、自动重试以及可选的进度追踪功能。 有关常规上传 API 概念(上传 URL、标头、分片),请参阅本地和远程上传的工作原理


先决条件

1

身份验证

您拥有一个可用的 Frameio 客户端。请参阅身份验证指南以了解设置流程。

2

安装 SDK

$ pip install frameio
3

目标文件夹

您需要知道文件要上传到的 account_idfolder_id。使用 SDK 来查找它们:

1 # List your accounts
2 accounts = client.accounts.index()
3 account_id = accounts.data[0].id
4
5 # List workspaces in the account
6 workspaces = client.workspaces.index(account_id=account_id)
7 workspace_id = workspaces.data[0].id
8
9 # List projects in the workspace
10 projects = client.projects.index(account_id=account_id, workspace_id=workspace_id)
11 project = projects.data[0]
12
13 # The project's root folder is the top-level upload target
14 folder_id = project.root_folder_id
15
16 # Or list subfolders to upload into a specific one
17 folders = client.folders.list(account_id=account_id, folder_id=folder_id)

快速入门

1import os
2from frameio import Frameio
3from frameio.files import FileCreateLocalUploadParamsData
4from frameio.upload import FrameioUploader
5
6client = Frameio(token="YOUR_TOKEN")
7
8file_path = "/path/to/video.mp4"
9file_size = os.path.getsize(file_path)
10
11# 1. Create the file resource and get pre-signed upload URLs
12response = client.files.create_local_upload(
13 account_id="YOUR_ACCOUNT_ID",
14 folder_id="YOUR_FOLDER_ID",
15 data=FileCreateLocalUploadParamsData(
16 name="video.mp4",
17 file_size=file_size,
18 ),
19)
20
21# 2. Upload the file to S3
22with open(file_path, "rb") as f:
23 FrameioUploader(response.data, f).upload()

就是这样。 SDK 根据 API 返回的上传 URL 将文件拆分为几个分片再并行上传,并自动处理重试。


工作原理

本地上传是一个两步过程:

1

创建文件资源

使用文件名和大小调用 client.files.create_local_upload()。 API 创建占位符文件并返回预签名的 S3 PUT URL(每个分片一个)。 URL 的数量(以及因此的分片数量)取决于具体的文件大小。

2

上传至 S3

FrameioUploader 从响应中读取上传 URL,将您的文件拆分为匹配的分片,并使用线程池将每个分片并行 PUT 到其 URL。 每个请求包含必需的 x-amz-acl: privateContent-Type 标头。

上传直接从您的应用程序传输到 S3(不通过 Frame.io API 服务器)。 这与 YouTube、Vimeo 和 Dropbox 等服务用于大文件上传的阵列相同。


FrameioUploader

FrameioUploader 是上传文件的推荐方式。 其中封装了底层分片的上传器并处理所有细节 — 从 API 响应中提取上传 URL、设置所需的标头、对文件分片以及并行上传。

进度跟踪

使用 on_progress 回调来跟踪上传进度:

1def on_progress(bytes_uploaded: int, total_bytes: int) -> None:
2 pct = bytes_uploaded / total_bytes * 100
3 print(f"\r{pct:.1f}% ({bytes_uploaded:,} / {total_bytes:,} bytes)", end="", flush=True)
4
5with open(file_path, "rb") as f:
6 FrameioUploader(response.data, f, on_progress=on_progress).upload()
7
8print("\nUpload complete!")

每个分片上传完成后,回调函数会被调用一次,返回到目前为止已上传的累计字节数和文件总大小。

丰富的进度条

为获得流畅的终端体验,请使用 Rich

1from rich.progress import Progress, BarColumn, DownloadColumn, TransferSpeedColumn, TimeRemainingColumn
2
3with Progress(
4 "[progress.description]{task.description}",
5 BarColumn(),
6 DownloadColumn(),
7 TransferSpeedColumn(),
8 TimeRemainingColumn(),
9) as progress:
10 task = progress.add_task("Uploading...", total=file_size)
11
12 with open(file_path, "rb") as f:
13 FrameioUploader(
14 response.data, f,
15 on_progress=lambda done, total: progress.update(task, completed=done),
16 ).upload()

配置

FrameioUploader 接受多个可选参数:

参数默认描述
max_workers5并发上传线程数
标头{"x-amz-acl": "private"}随每个 S3 PUT 请求发送的标头。 自定义标头与默认标头合并。
max_retries3每个分片的重试次数(指数退避:1 秒、2 秒、4 秒……)
on_progress在每个分片后触发回调 (bytes_uploaded, total_bytes)
1with open(file_path, "rb") as f:
2 FrameioUploader(
3 response.data,
4 f,
5 max_workers=10, # more parallelism for high-bandwidth connections
6 max_retries=5, # more resilient on flaky networks
7 on_progress=on_progress,
8 ).upload()

完整示例

包含身份验证、上传和进度跟踪的完整示例:

1import os
2from frameio import Frameio
3from frameio.auth import ServerToServerAuth
4from frameio.files import FileCreateLocalUploadParamsData
5from frameio.upload import FrameioUploader
6
7# Authenticate
8auth = ServerToServerAuth(
9 client_id="YOUR_CLIENT_ID",
10 client_secret="YOUR_CLIENT_SECRET",
11)
12client = Frameio(token=auth.get_token)
13
14# Prepare the file
15file_path = "/path/to/video.mp4"
16file_name = os.path.basename(file_path)
17file_size = os.path.getsize(file_path)
18
19# Create the file resource
20response = client.files.create_local_upload(
21 account_id="YOUR_ACCOUNT_ID",
22 folder_id="YOUR_FOLDER_ID",
23 data=FileCreateLocalUploadParamsData(
24 name=file_name,
25 file_size=file_size,
26 ),
27)
28
29print(f"Uploading {file_name} ({file_size:,} bytes) in {len(response.data.upload_urls)} chunks...")
30
31# Upload with progress
32def on_progress(uploaded: int, total: int) -> None:
33 print(f"\r{uploaded / total:.0%}", end="", flush=True)
34
35with open(file_path, "rb") as f:
36 FrameioUploader(response.data, f, on_progress=on_progress).upload()
37
38print(f"\nDone! View at: {response.data.view_url}")

如果您需要完全控制上传过程(例如,手动处理分片、集成异步管道或自定义重试逻辑),请参阅本地和远程上传的工作原理,了解原始 API 流程和独立 Python 脚本示例。


远程上传

如果您的文件已可通过公共 URL 访问,请改用远程上传。 无需分片,Frame.io 会直接获取文件:

1from frameio.files import FileCreateRemoteUploadParamsData
2
3response = client.files.create_remote_upload(
4 account_id="YOUR_ACCOUNT_ID",
5 folder_id="YOUR_FOLDER_ID",
6 data=FileCreateRemoteUploadParamsData(
7 name="video.mp4",
8 source_url="https://example.com/video.mp4",
9 ),
10)
11print(f"File created: {response.data.id}")

远程上传目前有 50 GB 的文件大小限制。 对于超过 50 GB 的文件,请改用本地上传


检查上传状态

上传后,您可以验证文件是否已接收:

1status = client.files.show_file_upload_status(
2 account_id="YOUR_ACCOUNT_ID",
3 file_id=response.data.id,
4)
5print(f"Upload complete: {status.data.upload_complete}")