ApsaraDB RDS:Migrate full backups to the cloud

Option 1: ossbrowser (recommended for most cases)

1. Download ossbrowser.

2. On Windows x64: decompress oss-browser-win32-x64.zip and run oss-browser.exe.

3. Select Log On With AK, enter your AccessKeyId and AccessKeySecret, and click Log On.

   An AccessKey pair is used for identity verification. Keep your AccessKey confidential.

   

4. Click the destination bucket.

   

5. Click , select the backup file, and click Open.

Option 2: OSS console (for files smaller than 5 GB)

1. Log in to the OSS console.

2. Click Buckets, then the name of your bucket.

   

3. In the Objects list, click Upload Object.

   

4. Drag the backup file to the Files to Upload area, or click Select Files to browse.

   

5. Click Upload Object.

Option 3: OSS API multipart upload (for files larger than 5 GB)

Use the alibabacloud-oss-v2 Python SDK for multipart upload with resumable support. Install the dependency first:

pip install alibabacloud-oss-v2

Set the following environment variables before running the script:

# -*- coding: utf-8 -*-
"""
Alibaba Cloud OSS Python SDK v2
Dependency: pip install alibabacloud-oss-v2
"""

import os
import sys
from pathlib import Path
import alibabacloud_oss_v2 as oss
from alibabacloud_oss_v2 import exceptions as oss_ex


def get_client_from_env(region: str, endpoint: str | None = None) -> oss.Client:
    """
    Create a v2 client from environment variables.
    - Prioritize using Region (recommended), but also support custom Endpoints (optional).
    - Compatible with both AK and STS:
        * AK: Requires OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET
        * STS: Also requires OSS_SESSION_TOKEN (compatible with the old variable OSS_SECURITY_TOKEN)
    """
    # Compatibility: If the user uses the old variable OSS_SECURITY_TOKEN, map it to the v2 expected OSS_SESSION_TOKEN
    sec_token_legacy = os.getenv("OSS_SECURITY_TOKEN")
    if sec_token_legacy and not os.getenv("OSS_SESSION_TOKEN"):
        os.environ["OSS_SESSION_TOKEN"] = sec_token_legacy

    ak = os.getenv("OSS_ACCESS_KEY_ID")
    sk = os.getenv("OSS_ACCESS_KEY_SECRET")
    st = os.getenv("OSS_SESSION_TOKEN")  # STS Token (optional)

    if not (ak and sk):
        raise ValueError("No valid AK found. Set the OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET environment variables. "
                         "If using STS, also set OSS_SESSION_TOKEN (or the old name OSS_SECURITY_TOKEN).")

    # Indicate the type of credential used
    if st:
        print("STS Token (OSS_SESSION_TOKEN) detected. Using STS credentials.")
    else:
        print("No STS Token detected. Using AccessKey (AK) credentials.")

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # Basic network configuration
    cfg.region = region  # Example: 'cn-hangzhou'
    if endpoint:
        # Optional: Custom Endpoint (e.g., internal network, accelerated, dedicated domain)
        cfg.endpoint = endpoint

    # You can also add other configurations here, such as: cfg.use_accelerate_endpoint = True
    return oss.Client(cfg)


def resumable_upload_file_v2(
    client: oss.Client,
    bucket_name: str,
    object_key: str,
    file_path: str,
    part_size: int = 1 * 1024 * 1024,
    parallel_num: int = 4,
    checkpoint_dir: str | None = None,
):
    """
    Implement concurrent multipart upload with resumable upload.

    :param client: Initialized oss.Client
    :param bucket_name: Destination bucket name
    :param object_key: Destination object key (without bucket name)
    :param file_path: Full path of the local file
    :param part_size: Part size in bytes, default is 1 MB
    :param parallel_num: Number of concurrent upload threads, default is 4
    :param checkpoint_dir: Directory to store breakpoint information; if None, resumable upload is disabled
    """
    file_path = str(file_path)
    if not Path(file_path).exists():
        raise FileNotFoundError(f"Error: Local file not found. Check the file_path configuration: {file_path}")

    # Construct the Uploader; enable resumable upload based on whether checkpoint_dir is provided
    if checkpoint_dir:
        uploader = client.uploader(
            enable_checkpoint=True,
            checkpoint_dir=checkpoint_dir,
            part_size=part_size,
            parallel_num=parallel_num,
        )
    else:
        uploader = client.uploader(
            part_size=part_size,
            parallel_num=parallel_num,
        )

    print(f"Starting to upload file: {file_path}")
    print(f"Destination Bucket: {bucket_name}")
    print(f"Destination Object: {object_key}")
    print(f"Part size: {part_size} bytes, Concurrency: {parallel_num}")
    if checkpoint_dir:
        print(f"Resumable upload: Enabled (checkpoint_dir={checkpoint_dir})")
    else:
        print("Resumable upload: Disabled (set checkpoint_dir to enable)")

    # Execute the upload (Uploader automatically chooses between multi/single part concurrent upload based on size)
    result = uploader.upload_file(
        oss.PutObjectRequest(bucket=bucket_name, key=object_key),
        filepath=file_path,
    )

    print("-" * 30)
    print("File uploaded successfully!")
    print(f"HTTP Status: {result.status_code}")
    print(f"ETag: {result.etag}")
    print(f"Request ID: {result.request_id}")
    # CRC-64 checksum; v2 enables data validation by default
    print(f"CRC64: {result.hash_crc64}")
    print("-" * 30)


def main():
    # Before running the code example, make sure you have set the corresponding environment variables.
    # macOS/Linux:
    #   AK method:
    #     export OSS_ACCESS_KEY_ID=YOUR_AK_ID
    #     export OSS_ACCESS_KEY_SECRET=YOUR_AK_SECRET
    #   STS method:
    #     export OSS_ACCESS_KEY_ID=YOUR_STS_ID
    #     export OSS_ACCESS_KEY_SECRET=YOUR_STS_SECRET
    #     export OSS_SECURITY_TOKEN=YOUR_STS_TOKEN
    #
    # Windows:
    #   Powershell: $env:OSS_ACCESS_KEY_ID="YOUR_AK_ID"
    #   cmd: set OSS_ACCESS_KEY_ID=YOUR_AK_ID

    # ===================== Parameters (modify as needed) =====================
    # Region example: 'cn-hangzhou'; we recommend using Region first
    region = "cn-hangzhou"

    # Optional: Custom Endpoint (for internal network, dedicated domain, accelerated domain name, etc.)
    # Example: 'https://oss-cn-hangzhou.aliyuncs.com'
    endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'

    # Bucket and Object
    bucket_name = "examplebucket"
    object_key = "test.bak"

    # Full path of the local file to upload.
    # Windows example: r'D:\localpath\examplefile.txt'  (note the r at the beginning)
    # macOS/Linux example: '/Users/test/examplefile.txt'
    file_path = r"D:\oss\test.bak"

    # Sharding and concurrency
    part_size = 1 * 1024 * 1024  # Default is 1 MB; OSS requires a minimum part size of 100 KB
    parallel_num = 4

    # Resumable upload directory (pass None to disable; we recommend specifying a writable directory)
    checkpoint_dir = str(Path.cwd() / ".oss_checkpoints")
    # =================== End of parameters ===================

    print("Script execution starts...")
    try:
        client = get_client_from_env(region=region, endpoint=endpoint)
        # If resumable upload is enabled, make sure the directory exists
        if checkpoint_dir:
            Path(checkpoint_dir).mkdir(parents=True, exist_ok=True)

        resumable_upload_file_v2(
            client=client,
            bucket_name=bucket_name,
            object_key=object_key,
            file_path=file_path,
            part_size=part_size,
            parallel_num=parallel_num,
            checkpoint_dir=checkpoint_dir,
        )
    except FileNotFoundError as e:
        print(e)
    except oss_ex.ServiceError as e:
        # Error returned by the OSS server
        print("\nAn OSS server-side error occurred.")
        print(f"HTTP Status: {getattr(e, 'status_code', 'N/A')}")
        print(f"Error Code: {getattr(e, 'code', 'N/A')}")
        print(f"Message: {getattr(e, 'message', 'N/A')}")
        print(f"Request ID: {getattr(e, 'request_id', 'N/A')}")
        print(f"Endpoint: {getattr(e, 'request_target', 'N/A')}")
    except oss_ex.BaseError as e:
        # SDK local/serialization/deserialization/credential errors
        print("\nAn OSS SDK client-side error occurred.")
        print(str(e))
    except Exception as e:
        print(f"\nAn unknown error occurred: {e}")


if __name__ == "__main__":
    main()