How to copy a file from one APS OSS bucket to another without having to download the file locally?

Is there a way to copy file from one APS (Forge) OSS bucket to another without needing to download the file locally? There is a solution posted Move files around on Forge, but it involves using Design Automation and has extra cost implications.

Is there another way?

Answering my own question here.

One can write a simple python script that calls request.get() on downloadable signed url using stream=True and call request.post() on the uploadable signed url. This will work for small files. For large files that need multipart upload, one would need to split the input stream into chunks and individually call on the multipart urls.

Here is the python code that works for small and large files both:

import json
import requests
import math

def getSourceUrl(token, bucket, object):
  url = f'https://developer.api.autodesk.com/oss/v2/buckets/{bucket}/objects/{object}/signed'
  
  payload = json.dumps({})
  headers = {
    'Authorization': f'Bearer {token}',
    'Content-Type': 'application/json'
  }
  response = requests.post(url, headers=headers, data=payload)
  return response.json()['signedUrl']

def getTargetUrlData(token, bucket, object, parts):
  url = f'https://developer.api.autodesk.com/oss/v2/buckets/{bucket}/objects/{object}/signeds3upload?parts={parts}'
  headers = {
    'Authorization': f'Bearer {token}',
    'Content-Type': 'application/json'
  }
  response = requests.get(url, headers=headers).json()
  
  return {
      'urls': response['urls'],
      'uploadKey': response['uploadKey']
  }

def getAccessToken(clientId, clientSecret):
  url   = 'https://developer.api.autodesk.com/authentication/v1/authenticate'
  payload = f'client_id={clientId}&client_secret={clientSecret}&grant_type=client_credentials&scope=data:read data:write'
  headers = { 'Content-Type': 'application/x-www-form-urlencoded' }
  response = requests.post(url, headers=headers, data=payload).json()

  return response['access_token']

def copyFromSourceToTarget(sourceData, targetData):
  sourceToken = getAccessToken(sourceData['clientId'], sourceData['clientSecret'])
  targetToken = getAccessToken(targetData['clientId'], targetData['clientSecret'])

  sourceUrl = getSourceUrl(sourceToken, sourceData['bucket'], sourceData['object'])
  sourceResponse = requests.get(sourceUrl, headers={'Accept-Encoding': None}, stream=True)

  chunksSize = 5 * 1024 * 1024
  chunksNum  = math.ceil(int(sourceResponse.headers['content-length']) / chunksSize)
  targetUrls = getTargetUrlData(targetToken, targetData['bucket'], targetData['object'], chunksNum)

  count = 0
  for chunk in sourceResponse.iter_content(chunksSize):
    headers = { 'Content-Type': 'application/octet-stream', 'Content-Length': f'{chunksSize}' }
    res = requests.put(targetUrls['urls'][count], headers=headers, data = chunk)
    count += 1

  headers = { 'Content-Type': 'application/json', 'Authorization': f'Bearer {targetToken}', 'x-ads-meta-Content-Type': 'application/octet-stream'}
  payload = json.dumps({'uploadKey': targetUrls['uploadKey']})
  res = requests.post(f'https://developer.api.autodesk.com/oss/v2/buckets/{targetData["bucket"]}/objects/{targetData["object"]}/signeds3upload', headers=headers, data = payload)

  return res.json()

You can then call this simply with source and target file information. The APS app (client id / secret) may or mayn't be the same for source and target. The code takes care of both.

def my_handler(event, context):
  sourceData = {
    'clientId': '<source client id>',
    'clientSecret': '<source client secret>',
    'bucket': 'bucket1',
    'object': 'source.rvt'
  }
  targetData = {
    'clientId': '<target client id>',
    'clientSecret': '<target client secret>',
    'bucket': 'bucket2',
    'object': 'target.rvt'
  }

  return copyFromSourceToTarget(sourceData, targetData)