ApsaraDB RDS:フルバックアップのクラウド移行

オプション 1：ossbrowser（ほとんどの場合に推奨）

1. ossbrowser をダウンロードします。

2. Windows x64 環境の場合：oss-browser-win32-x64.zip を解凍し、oss-browser.exe を実行します。

3. AK でログイン を選択し、AccessKeyId および AccessKeySecret を入力して、「ログイン」をクリックします。

   AccessKey ペアは本人確認に使用されます。AccessKey は機密情報として厳重に管理してください。

   

4. 送信先バケットをクリックします。

   

5. をクリックし、バックアップファイルを選択して、「開く」をクリックします。

オプション 2：OSS コンソール（5 GB 未満のファイルに推奨）

1. OSS コンソール にログインします。

2. バケット をクリックし、バケット名をクリックします。

   

3. オブジェクト リストで、「オブジェクトのアップロード」をクリックします。

   

4. バックアップファイルを アップロードするファイル エリアにドラッグするか、「ファイルの選択」をクリックしてファイルを参照します。

   

5. オブジェクトのアップロード をクリックします。

オプション 3：OSS API マルチパートアップロード（5 GB 超のファイルに推奨）

再開可能なアップロード機能付きの alibabacloud-oss-v2 Python SDK を使用してマルチパートアップロードを行います。まず依存関係をインストールします。

pip install alibabacloud-oss-v2

スクリプトを実行する前に、以下の環境変数を設定します。

# -*- 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()