Move Files Between Different Cloud Storage Services

Getting Started

Basic Concept

Transfer files and objects across cloud services, accounts, and regions

Files and objects may be stored in different environments depending on the cloud service, account, project, or region.

Cloud-to-cloud storage transfer moves files and objects from a source storage environment to a designated target and creates a data-transfer path across the required accounts and regions.

Source Storage
      │
      ▼
Transfer Flow
      │
      ▼
Target Storage

For example, you can transfer from Amazon S3 to Azure Blob or connect different accounts and regions within the same cloud service.

Transfer Flow

Connect the process from source selection through delivery to the target storage

Cloud-to-cloud transfer starts by connecting the source and target storage environments, selecting the files to transfer, and applying them to the configured destination path.

Source Storage
      │
      ▼
Connect Account · Region
      │
      ▼
Set Transfer Path
      │
      ▼
Transfer Files · Objects
      │
      ▼
Apply to Target Storage

Benefits

Move distributed cloud data to the environment where it is needed

Cloud-to-cloud storage transfer lets you move data distributed across services, accounts, and regions to storage locations that match your business and operational requirements.

IT Engineers

Configure and operate large-file transfers across cloud storage environments

Connect Storage

Configure the cloud environments and access scope used for transfer

First, connect the cloud storage services you will use, including Amazon S3, Azure Blob, GCS, and Cloudflare R2.

Configure each storage account, project, region, and access scope to prepare the environments used by transfer jobs.

Amazon S3 ──────┐
Azure Blob ─────┤
GCS ────────────┼──→ Transfer Environment
Cloudflare R2 ──┘

At this stage, configure the transfer environment and access scope together, combining the previous Storage Connection and Account · Region concepts into one step.

Transfer Path

Connect the source bucket to the target storage location

Specify the source bucket or container and the target storage location, then configure the file-movement path.

When needed, a single transfer job can connect storage locations across different services, accounts, and regions.

Source
Account A / Region 1
Bucket: media-source
          │
          ▼
      Transfer
          │
          ▼
Target
Account B / Region 2
Bucket: media-archive

Transfer Rules

Configure transfer jobs according to file scale and processing conditions

When transferring large files or many objects, configure both the file scope and execution conditions.

Select transfer targets based on file path, type, name, and similar criteria, then configure the job to run on a schedule, when files are created, or in response to an external request.

Source Storage
       │
       ▼
   Transfer Rules
       │
 ┌─────┼──────────┐
 ▼                ▼
Files          Objects
 │                │
 └──────┬─────────┘
        ▼
   Transfer Run
        │
        ▼
 Target Storage

The previous Bulk Transfer and Transfer Conditions sections share the same purpose of defining how the actual transfer job runs, so they are combined into one step.

Verify Results

Review transferred files and processing status by target

When a transfer runs, use Runs to review the source and target storage, progress, processed file count, total size, and execution status.

Transfer Run
     │
     ├── Source / Target
     │
     ├── Trigger
     │
     ├── Progress
     │
     └── Status

Reviewing the execution result for each job lets you manage whether files and objects were applied successfully to each target storage environment.

Operations Management

Manage transfers across multiple cloud environments from one place

When using multiple cloud services, accounts, and regions, you can review the execution history and processing status of transfer jobs together.

If a specific transfer requires additional review, inspect the detailed execution result and rerun the necessary work.

             Cloud Operations
                    │
       ┌────────────┼────────────┐
       ▼            ▼            ▼
   S3 Transfer  Azure Transfer  GCS Transfer
       │            │            │
       └────────────┼────────────┘
                    ▼
             Runs & Results

Developers

Transfer objects between different cloud storage services and verify the result by comparing object counts

Integration Setup

Prepare shared API calls and status values

import os
import requests

BASE_URL = os.getenv("INNORIX_BASE_URL", "https://app.innorix.com").rstrip("/")
TOKEN = os.environ["INNORIX_ACCESS_TOKEN"]
WORKSPACE_ID = os.getenv("INNORIX_WORKSPACE_ID")   # optional; falls back to the current workspace

STATUS_COMPLETE = 2
TERMINAL = {2, 4, 5, 9, 99}          # complete / error / cancelled / partial / failed
NOT_SUCCEEDED = {4, 5, 9, 99}


def api(method, path, body=None, params=None):
    headers = {
        "Content-Type": "application/json",
        "Authorization": f"Bearer {TOKEN}",
    }

    if WORKSPACE_ID:
        headers["x-workspace-id"] = WORKSPACE_ID

    response = requests.request(
        method, BASE_URL + path,
        headers=headers, json=body, params=params, timeout=30,
    )

    payload = response.json() if response.content else {}

    if not response.ok:
        raise RuntimeError(payload.get("message") or f"HTTP {response.status_code}")

    return payload.get("data")


def is_terminal(detail):
    return detail.get("isTerminal", detail.get("status") in TERMINAL)

Use the following values to determine transfer status. There are five terminal states, and Complete (2) is the successful state.

Select Storage

Identify the device IDs and paths for connected storage environments

Connected cloud storage appears as transfer devices. Retrieve the identifiers from the device list.

result = api("GET", "/api/devices", params={"page": 1, "size": 200}) or {}

for device in result.get("devices") or []:
    print(device["deviceId"], device["name"], device.get("os"))

The device list is returned in the data.devices array.

If you work by name, resolve the name to an identifier first. When several connections use the same cloud service across different accounts or regions, similar names can be confusing.

def resolve_device(name):
    result = api("GET", "/api/devices/resolve", params={"name": name}) or {}
    devices = result.get("devices") or []
    count = result.get("matchCount", len(devices))

    if count != 1 or not devices:
        raise RuntimeError(f"{name}: {count} matches - use a more specific name")

    return devices[0]["deviceId"]

Use the bucket or container path as the prefix directly. Storage services accept prefix-style paths, so no additional transformation is required.

Run Transfer

Send objects under the source prefix to the target

def move_objects(source, target, source_prefix, target_prefix):
    transfer = api("POST", "/api/transfers/manual", {
        "sourceDevice": source,
        "targetDevice": target,
        "targetPath": target_prefix,
        "sourcePaths": [source_prefix],
        "sendAllFolder": True,
        "transferOptions": {"target-action": "overwrite"},
    })

    return transfer["monitorId"]

For service-to-service movement, overwrite is the safer option. Using numbering changes object keys and can break paths referenced by applications.

To review transfer targets in advance, list the source storage contents.

result = api("GET", f"/api/devices/{SOURCE}/files", params={
    "path": "media/2026/09",
    "page": 1, "size": 200, "type": "file",
})

print(result.get("total"), "objects")

if result.get("truncated"):
    print("result was truncated - narrow the prefix")

Process Large Object Sets

Split objects by prefix and move them in multiple transfers

If a bucket contains millions of objects, moving everything at once can be impractical. Splitting by prefix allows only the failed range to be retransferred.

PREFIXES = [f"media/2026/{month:02d}" for month in range(1, 13)]

transfers = {
    prefix: move_objects(SOURCE, TARGET, prefix,
                         prefix.replace("media", "archive", 1))
    for prefix in PREFIXES
}

def wait(monitor_id, timeout=3600, interval=3):
    deadline = time.time() + timeout

    while time.time() < deadline:
        detail = api("GET", f"/api/transfers/{monitor_id}")

        if is_terminal(detail):
            return detail

        time.sleep(interval)

    raise TimeoutError(monitor_id)


def failed_files(monitor_id):
    result = api("GET", f"/api/transfers/{monitor_id}/files", params={
        "state": "any", "size": 500,
    }) or {}

    return [r for r in (result.get("children") or [])
            if r.get("status") in NOT_SUCCEEDED]


def retry_failed(monitor_id):
    rows = failed_files(monitor_id)

    if not rows:
        return 0

    api("POST", f"/api/transfers/{monitor_id}/retry", {
        "filesRetry": [
            {"filePath": r["sourceFilePath"], "isDir": bool(r.get("isFolder"))}
            for r in rows
        ]
    })

    return len(rows)

for prefix, monitor_id in transfers.items():
    detail = wait(monitor_id, timeout=14400)

    print(f"{prefix:24} {detail.get('statusName')}"
          f" {detail.get('fileCount')} objects {detail.get('totalSize')} bytes")

    if detail["status"] != STATUS_COMPLETE:
        print(f"  retried {retry_failed(monitor_id)} files")

Continue reviewing the remaining ranges even if one fails so you can determine how far the migration completed.

Conditional Transfer

Select objects to move by type and size

Often, only specific files need to be moved rather than the entire dataset.

def build_filter(exts=None, min_size=None, exclude=None):
    file_option = {}

    if exts:
        # extension whitelist, without the leading dot
        file_option["extension"] = {
            "extension": [e.lstrip(".").lower() for e in exts],
            "allow": True,
        }

    if min_size is not None:
        # over and equal both True means size or larger
        file_option["fileSize"] = {"size": min_size, "over": True, "equal": True}

    if exclude:
        # allow=False excludes files whose name contains this. Server matching is case sensitive.
        file_option["fileName"] = {"name": exclude, "allow": False}

    return {"send-fileoption": file_option} if file_option else {}

Use send-fileoption.extension for extension filters. The send-filetype-cus regular expression matches only the file name without the extension, so it does not work as an extension filter.

When multiple filters are provided, they are combined with AND. Only files that match every condition are transferred.

options = {
    **build_filter(exts=["mp4", "mov"], min_size=1048576),
    "target-action": "overwrite",
}

api("POST", "/api/transfers/manual", {
    "sourceDevice": SOURCE,
    "targetDevice": TARGET,
    "targetPath": "archive/2026/09",
    "sourcePaths": ["media/2026/09"],
    "sendAllFolder": True,
    "transferOptions": options,
})

Incremental Transfer

Send only objects added or modified since the last transfer

If the same prefix is moved periodically, there is no need to retransmit the entire set every time.

transfer = api("POST", "/api/transfers/manual", {
    "sourceDevice": SOURCE,
    "targetDevice": TARGET,
    "targetPath": "archive/2026/09",
    "sourcePaths": ["media/2026/09"],
    "sendAllFolder": True,
    "incremental": True,
    "transferOptions": {"target-action": "overwrite"},
})

Incremental transfer is disabled by default. The agent calculates the delta and transfers only objects added or modified since the previous transfer. Always use overwrite with incremental transfer.

Verify Results

Compare source and target object counts

Unlike file systems, object storage may not support checksum retrieval. Compare the source and target object counts to confirm that nothing is missing.

def count_objects(device, prefix):
    total, page = 0, 1

    while True:
        result = api("GET", f"/api/devices/{device}/files", params={
            "path": prefix,
            "page": page, "size": 1000, "type": "file",
        }) or {}

        total += len(result.get("items") or [])

        if page >= (result.get("lastPage") or 1):
            return total

        page += 1


source_count = count_objects(SOURCE, "media/2026/09")
target_count = count_objects(TARGET, "archive/2026/09")

if source_count != target_count:
    raise RuntimeError(
        f"object counts differ: source {source_count} / target {target_count}")

If the counts differ, the transfer is incomplete. Retransfer the failed prefix ranges until the counts match.