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
依存関係: 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:
    """
    環境変数から v2 クライアントを作成します。
    - リージョンの使用を優先しますが (推奨)、カスタムエンドポイントもサポートします (オプション)。
    - AK と STS の両方に対応：
        * AK: OSS_ACCESS_KEY_ID / OSS_ACCESS_KEY_SECRET が必要
        * STS: OSS_SESSION_TOKEN も必要 (旧変数 OSS_SECURITY_TOKEN と互換性あり)
    """
    # 互換性：ユーザーが旧変数 OSS_SECURITY_TOKEN を使用している場合、v2 が期待する 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 トークン (オプション)

    if not (ak and sk):
        raise ValueError("有効な AK が見つかりません。環境変数 OSS_ACCESS_KEY_ID と OSS_ACCESS_KEY_SECRET を設定してください。"
                         "STS を使用する場合は、OSS_SESSION_TOKEN (または旧名の OSS_SECURITY_TOKEN) も設定してください。")

    # 使用する認証情報の種類を示します
    if st:
        print("STS トークン (OSS_SESSION_TOKEN) が検出されました。STS 認証情報を使用します。")
    else:
        print("STS トークンが検出されませんでした。AccessKey (AK) 認証情報を使用します。")

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

    # 基本的なネットワーク構成
    cfg.region = region  # 例：'cn-hangzhou'
    if endpoint:
        # オプション：カスタムエンドポイント (内部ネットワーク、高速化、専用ドメインなど)
        cfg.endpoint = endpoint

    # ここに他の構成を追加することもできます。例：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,
):
    """
    再開可能なアップロードを伴う同時マルチパートアップロードを実装します。

    :param client: 初期化された oss.Client
    :param bucket_name: 移行先バケット名
    :param object_key: 移行先オブジェクトキー (バケット名なし)
    :param file_path: ローカルファイルのフルパス
    :param part_size: パートサイズ (バイト単位)、デフォルトは 1 MB
    :param parallel_num: 同時アップロードスレッド数、デフォルトは 4
    :param checkpoint_dir: ブレークポイント情報を保存するディレクトリ。None の場合、再開可能なアップロードは無効になります
    """
    file_path = str(file_path)
    if not Path(file_path).exists():
        raise FileNotFoundError(f"エラー：ローカルファイルが見つかりません。file_path の構成を確認してください：{file_path}")

    # アップローダーを構築します。checkpoint_dir が提供されているかどうかに基づいて、再開可能なアップロードを有効にします
    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"ファイルのアップロードを開始します：{file_path}")
    print(f"移行先バケット：{bucket_name}")
    print(f"移行先オブジェクト：{object_key}")
    print(f"パートサイズ：{part_size} バイト、同時実行数：{parallel_num}")
    if checkpoint_dir:
        print(f"再開可能なアップロード：有効 (checkpoint_dir={checkpoint_dir})")
    else:
        print("再開可能なアップロード：無効 (有効にするには checkpoint_dir を設定してください)")

    # アップロードを実行します (アップローダーはサイズに基づいてマルチ/シングルパートの同時アップロードを自動的に選択します)
    result = uploader.upload_file(
        oss.PutObjectRequest(bucket=bucket_name, key=object_key),
        filepath=file_path,
    )

    print("-" * 30)
    print("ファイルが正常にアップロードされました！")
    print(f"HTTP ステータス：{result.status_code}")
    print(f"ETag: {result.etag}")
    print(f"リクエスト ID: {result.request_id}")
    # CRC-64 チェックサム。v2 はデフォルトでデータ検証を有効にします
    print(f"CRC64: {result.hash_crc64}")
    print("-" * 30)


def main():
    # コード例を実行する前に、対応する環境変数を設定していることを確認してください。
    # macOS/Linux:
    #   AK メソッド：
    #     export OSS_ACCESS_KEY_ID=YOUR_AK_ID
    #     export OSS_ACCESS_KEY_SECRET=YOUR_AK_SECRET
    #   STS メソッド：
    #     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

    # ===================== パラメーター (必要に応じて変更) =====================
    # リージョンの例：'cn-hangzhou'。まずリージョンを使用することを推奨します
    region = "cn-hangzhou"

    # オプション：カスタムエンドポイント (内部ネットワーク、専用ドメイン、高速化ドメイン名など)
    # 例：'https://oss-cn-hangzhou.aliyuncs.com'
    endpoint = 'https://oss-cn-hangzhou.aliyuncs.com'

    # バケットとオブジェクト
    bucket_name = "examplebucket"
    object_key = "test.bak"

    # アップロードするローカルファイルのフルパス。
    # Windows の例：r'D:\localpath\examplefile.txt' (先頭の r に注意)
    # macOS/Linux の例：'/Users/test/examplefile.txt'
    file_path = r"D:\oss\test.bak"

    # 分割と同時実行
    part_size = 1 * 1024 * 1024  # デフォルトは 1 MB。OSS は最小パートサイズ 100 KB を要求します
    parallel_num = 4

    # 再開可能なアップロードディレクトリ (無効にするには None を渡します。書き込み可能なディレクトリを指定することを推奨します)
    checkpoint_dir = str(Path.cwd() / ".oss_checkpoints")
    # =================== パラメーターの終わり ===================

    print("スクリプトの実行を開始します...")
    try:
        client = get_client_from_env(region=region, endpoint=endpoint)
        # 再開可能なアップロードが有効な場合、ディレクトリが存在することを確認します
        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:
        # OSS サーバーから返されたエラー
        print("\nOSS サーバー側でエラーが発生しました。")
        print(f"HTTP ステータス：{getattr(e, 'status_code', 'N/A')}")
        print(f"エラーコード：{getattr(e, 'code', 'N/A')}")
        print(f"メッセージ：{getattr(e, 'message', 'N/A')}")
        print(f"リクエスト ID: {getattr(e, 'request_id', 'N/A')}")
        print(f"エンドポイント：{getattr(e, 'request_target', 'N/A')}")
    except oss_ex.BaseError as e:
        # SDK のローカル/シリアル化/デシリアル化/認証情報エラー
        print("\nOSS SDK クライアント側でエラーが発生しました。")
        print(str(e))
    except Exception as e:
        print(f"\n不明なエラーが発生しました：{e}")


if __name__ == "__main__":
    main()