Hi @freddy you’re going to want to use AWS multi-part upload for uploading to Frame.io. When you make the call to Frame.io for a local upload the server will send back a series of URLs for you to upload to which means you then need to chunk your file for the amount of URLs that are returned and then upload those chunks to those URLs. Frame will assemble them all together in the background so you dont have to worry about that part.
While we don’t yet have the upload logic in our V4 SDK, here’s some pseudo code using our SDK to help:
import os
import time
import mimetypes
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from frameio import Frameio
def build_retrying_session() -> requests.Session:
session = requests.Session()
# Retries for transient failures (network hiccups, 5xx, throttling, etc.)
retry = Retry(
total=8,
connect=8,
read=8,
backoff_factor=0.75,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["PUT"],
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry, pool_connections=8, pool_maxsize=8)
session.mount("https://", adapter)
return session
def upload_local_file_v4(
*,
client: Frameio,
account_id: str,
folder_id: str,
file_path: str,
poll_interval_s: int = 5,
) -> str:
file_size = os.path.getsize(file_path)
file_name = os.path.basename(file_path)
# 1) Create placeholder file + get presigned upload_urls
create_resp = client.files.create_local_upload(
account_id=account_id,
folder_id=folder_id,
data={"file_size": file_size, "name": file_name},
)
# SDK typically returns a dict-like object with "data"
file_obj = create_resp["data"] if isinstance(create_resp, dict) else create_resp.data
file_id = file_obj["id"]
media_type = file_obj.get("media_type") or (mimetypes.guess_type(file_path)[0] or "application/octet-stream")
upload_urls = file_obj["upload_urls"] # list of { "size": int, "url": str }
# 2) Upload each part (IMPORTANT: use the per-part size returned by the API)
session = build_retrying_session()
# Timeouts: (connect_timeout, read_timeout). Tune read_timeout for slow links.
timeout = (10, 300)
with open(file_path, "rb") as f:
for idx, part in enumerate(upload_urls, start=1):
part_size = int(part["size"])
url = part["url"]
chunk = f.read(part_size)
if len(chunk) != part_size:
raise RuntimeError(
f"Read {len(chunk)} bytes for part {idx}, expected {part_size}. "
"File changed during upload or sizing mismatch."
)
resp = session.put(
url,
data=chunk, # bytes to avoid chunked transfer encoding surprises
headers={
"Content-Type": media_type,
"x-amz-acl": "private",
},
timeout=timeout,
)
if resp.status_code != 200:
raise RuntimeError(
f"Part {idx} upload failed: HTTP {resp.status_code} - {resp.text}"
)
# 3) Poll for completion (Frame stitches + processes automatically)
while True:
status_resp = client.files.show_file_upload_status(account_id=account_id, file_id=file_id)
status_obj = status_resp["data"] if isinstance(status_resp, dict) else status_resp.data
if status_obj.get("upload_failed"):
raise RuntimeError(f"Frame.io reports upload_failed=true for file_id={file_id}")
if status_obj.get("upload_complete"):
return file_id
time.sleep(poll_interval_s)
# Example usage:
# client = Frameio(base_url="https://api.frame.io", token="YOUR_ACCESS_TOKEN")
# file_id = upload_local_file_v4(
# client=client,
# account_id="...",
# folder_id="...",
# file_path="/mnt/nas/huge_video.mov",
# )
# print("Upload complete:", file_id)
If you’re using AI to help you code, you can use our “Open in ChatGPT” function on our developer docs: Create file (local upload) | Frame.io API Documentation
We don’t have an uploader CLI at this time nor do we have auto-upload from Transfer just yet (we do have the MacOS app on the app store that has a “watch folder” style upload if you want to use that instead.
