使用 Python SDK 自動化遷移任務 - 資料線上遷移

本文介紹如何使用SDK調用任務相關的方法。

建立任務

說明

該方法主要用於建立遷移任務。
建立任務完成後，可以調用更新任務來啟動任務。

以下範例程式碼用於建立任務。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import ImportQos, FileTypeFilters, FilterRule, LastModifiedFilters, \
    LastModifyFilterItem, TimeFilter, KeyFilters, KeyFilterItem, ScheduleRule, CreateJobRequest, CreateJobInfo
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)
        # 填寫任務名稱。
        job_name = "examplejob"
        # 填寫源端資料位址名稱。
        src_address = "examplesrcaddress"
        # 填寫目的端資料位址名稱。
        dest_address = "exampledestaddress"

        """
            overwriteMode和transferMode需要組合使用，具體組合含義如下
            always, all
            全覆蓋
            always, lastmodified
            根據檔案最後修改時間覆蓋
            never, ""
            不覆蓋
        """
        """
            檔案覆蓋方式, 可能的值
            1.never
            2.always
        """
        overwrite_mode = "always"
        """
            檔案傳輸的方式, 可能的值為
            1.changed
            2.all
            3.lastmodified
        """
        transfer_mode = "lastmodified"

        # maxBandWidth，maxImportTaskQps根據實際需求值填寫。
        max_import_task_qps = 1000
        max_band_width = 2147483648
        # 如果maxImportTaskQps值為0或者不設定，會被設定為預設值;MaxBandWidth值為0或者不設定, 會被設定為預設值，maxBandWidth值單位為bits。
        import_qos = ImportQos(max_import_task_qps=max_import_task_qps, max_band_width=max_band_width)

        # 配置過濾規則，包含檔案類型過濾器、檔案過濾器、時間過濾器，具體參數含義請參看API文檔。
        # 檔案類型過濾器，適用於localfs。
        exclude_symlink = False
        exclude_dir = False
        file_type_filters = FileTypeFilters(exclude_dir=exclude_dir, exclude_symlink=exclude_symlink)
        # 檔案過濾器，根據實際需求值填寫。
        include_regex = [".*.jpg", ".*.gif"]
        exclude_regex = ["txtFile", ".*.js"]
        exclude_regex_item = KeyFilterItem(regex=exclude_regex)
        include_regex_item = KeyFilterItem(regex=include_regex)
        Key_filters = KeyFilters(excludes=exclude_regex_item, includes=include_regex_item)
        # 時間過濾器, 時間格式遵循UTC時間格式，根據實際需求值填寫。
        include_start_time = "2006-01-01T00:00:00Z"
        include_end_time = "2007-01-01T00:00:00Z"
        exclude_start_time = "2009-01-01T00:00:00Z"
        exclude_end_time = "2010-01-01T00:00:00Z"
        include_time_filters = [TimeFilter(start_time=include_start_time, end_time=include_end_time)]
        exculde_time_filters = [TimeFilter(start_time=exclude_start_time, end_time=exclude_end_time)]
        include_lastModify_filter_item = LastModifyFilterItem(time_filter=include_time_filters)
        exclude_lastModify_filter_tiem = LastModifyFilterItem(time_filter=exculde_time_filters)
        last_modified_filters = LastModifiedFilters(excludes=exclude_lastModify_filter_tiem,
                                                    includes=include_lastModify_filter_item)

        filterRule = FilterRule(file_type_filters=file_type_filters,
                                key_filters=Key_filters,
                                last_modified_filters=last_modified_filters)

        # 配置調度規則，具體參數含義請參看API文檔。
        max_schedule_count = 5
        start_cron_expression = "0 0 10 * * ?"
        suspend_cron_expression = "0 0 14 * * ?"
        schedule_rule = ScheduleRule(max_schedule_count=max_schedule_count,
                                     suspend_cron_expression=suspend_cron_expression,
                                     start_cron_expression=start_cron_expression)
        client.create_job(userid, CreateJobRequest(CreateJobInfo(
            name=job_name,
            transfer_mode=transfer_mode,
            overwrite_mode=overwrite_mode,
            src_address=src_address,
            dest_address=dest_address,
            import_qos=import_qos,
            filter_rule=filterRule,
            schedule_rule=schedule_rule,
        )))
    except Exception as e:
        print(e)

更新任務

說明

建立任務後，您可以通過該方法來啟動任務、暫停任務以及關閉任務，也可以通過該方法來調整任務限流資訊。

以下範例程式碼用於更新任務狀態。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import UpdateJobRequest, UpdateJobInfo, ImportQos
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)
        # 填寫任務名稱。
        job_name = "examplejob"
              
        # 允許的狀態值有IMPORT_JOB_LAUNCHING(啟動任務), IMPORT_JOB_SUSPEND（暫停任務）,IMPORT_JOB_CLOSING（關閉任務）。
        status = "IMPORT_JOB_LAUNCHING"
        client.update_job(userid, job_name, UpdateJobRequest(UpdateJobInfo(
            status=status
        )))
    except Exception as e:
        print(e)

以下範例程式碼用於更新任務限流。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import UpdateJobRequest, UpdateJobInfo, ImportQos
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)
        # 填寫任務名稱。
        job_name = "examplejob"
        # maxBandWidth、maxImportTaskQps請根據需求填寫。
        max_band_width = 1610612736
        max_import_task_qps = 1500
        client.update_job(userid, job_name, UpdateJobRequest(UpdateJobInfo(
            import_qos=ImportQos(
                max_band_width=max_band_width,
                max_import_task_qps=max_import_task_qps,
            )
        )))
    except Exception as e:
        print(e)

擷取任務詳情

說明

該方法主要用於擷取遷移任務詳情。

以下範例程式碼用於擷取特定任務的詳細資料（通過任務名稱）。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import GetJobRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)
        # 填寫任務名稱。
        job_name = "examplejob"
        response = client.get_job(userid, job_name, GetJobRequest())
        print(response.body.import_job)
    except Exception as e:
        print(e)

以下範例程式碼用於擷取指定任務的詳情（通過任務ID）。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import GetJobRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)
        # 填寫任務ID。
        version = "b4155550-****-4371-****-9c7337348021"
        response = client.get_job(userid, version, GetJobRequest(by_version=""))
        print(response.body.import_job)
    except Exception as e:
        print(e)

正常返回樣本

{
  "ImportJob": {
    "Name": "test_name",
    "SrcAddress": "test_src_address",
    "DestAddress": "test_dest_address",
    "Status": "IMPORT_JOB_DOING",
    "EnableMultiVersioning": false,
    "CreateTime": "2024-05-01T12:00:00.000Z",
    "ModifyTime": "2024-05-01T12:00:00.000Z",
    "Version": "test_id",
    "Audit": {
      "LogMode": "off"
    },
    "OverwriteMode": "always",
    "TransferMode": "all",
    "Tags": "K1:V1,K2:V2",
    "ParentName": "test_parent_name",
    "ParentVersion": "7db93837-a5ee-4e3a-b3c8-800e7947dabc",
    "ConvertSymlinkTarget": false,
    "CreateReport": false,
    "Owner": "test_owner",
    "FilterRule": {
      "KeyFilters": {
        "Includes": {
          "Regex": [
            ".*\\.jpg$"
          ]
        },
        "Excludes": {
          "Regex": [
            ".*\\.jpg$"
          ]
        }
      },
      "LastModifiedFilters": {
        "Includes": {
          "TimeFilter": [
            {
              "StartTime": "2006-01-01T00:00:00Z",
              "EndTime": "2006-12-31T59:59:59Z"
            }
          ]
        },
        "Excludes": {
          "TimeFilter": [
            {
              "StartTime": "2006-01-01T00:00:00Z",
              "EndTime": "2006-12-31T59:59:59Z"
            }
          ]
        }
      },
      "FileTypeFilters": {
        "ExcludeSymlink": true,
        "ExcludeDir": true
      }
    },
    "ImportQos": {
      "MaxBandWidth": 1073741824,
      "MaxImportTaskQps": 1000
    },
    "ScheduleRule": {
      "StartCronExpression": "0 0 * * * ?",
      "SuspendCronExpression": "0 0 * * * ?",
      "MaxScheduleCount": 1
    }
  }
}

列舉任務

說明

該方法主要用於列舉使用者在該地區下的遷移任務。

以下範例程式碼用於列舉帳號下所有任務。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import ListJobRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)
        # 根據實際填寫marker,count。
        count = 2
        marker = ""
        response = client.list_job(userid, ListJobRequest(
            marker=marker,
            count=2,
        ))
        for import_job in response.body.import_job_list.import_job:
            print(import_job)
    except Exception as e:
        print(e)

擷取任務重試資訊

說明

任務運行完成後，可能存在失敗檔案，線上遷移服務會對這些失敗檔案構造一份清單。通過以下範例程式碼擷取指定輪次失敗檔案的清單資訊，如返回結果為Ready，可用該清單重試失敗檔案。
該方法的結果主要用於重試任務的參數配置中。
該方法主要用於當遷移任務有失敗檔案時，擷取失敗檔案是否能重試的資訊。

以下範例程式碼用於擷取指定輪次的任務重試資訊。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import GetJobResultRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)

        # 填寫任務名稱。
        job_name = "examplejob"
        # 填寫任務輪次。
        runtime_id = 1
        response = client.get_job_result(userid, job_name, GetJobResultRequest(
            runtime_id=runtime_id
        ))
        print(response.body.import_job_result)
    except Exception as e:
        print(e)

正常返回樣本

{
  "ImportJobResult": {
    "ReadyRetry": "Ready",
    "InvPath": "mainfest.json",
    "InvBucket": "test_sys_bucket",
    "InvDomain": "test_domain",
    "InvLocation": "oss",
    "InvAccessId": "test_access_id",
    "InvAccessSecret": "test_secret_key",
    "InvRegionId": "test_region_id",
    "AddressType": "ossinv",
    "TotalObjectCount": 1000,
    "CopiedObjectCount": 800,
    "FailedObjectCount": 200,
    "SkippedObjectCount": 800,
    "TotalObjectSize": 1000,
    "CopiedObjectSize": 800,
    "SkippedObjectSize": 800,
    "Version": "test_job_id"
  }
}

列舉任務記錄

說明

該方法主要用於查詢指定任務記錄。

以下範例程式碼用於列舉指定任務的記錄。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import ListJobHistoryRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)

        # 填寫任務名稱。
        job_name = "examplejob"
        # 下列參數值請根據實際需求填寫。
        marker = ""
        count = 2
        response = client.list_job_history(userid, job_name, ListJobHistoryRequest(marker=marker, count=count))
        for job_history in response.body.job_history_list.job_history:
            print(job_history)
    except Exception as e:
        print(e)

刪除任務

說明

該方法主要用於刪除遷移任務。

以下範例程式碼用於刪除指定任務。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import DeleteJobRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)

        # 填寫任務名稱。
        job_name = "examplejob"
        client.delete_job(userid, job_name, DeleteJobRequest())
    except Exception as e:
        print(e)

建立任務遷移報告

說明

該方法為非同步方法呼叫，調用該方法後後端開始準備產生遷移報告，是否產生完畢需要通過擷取任務遷移報告方法來完成。

以下範例程式碼用於建立指定任務遷移報告。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import CreateReportRequest, CreateReportInfo
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)

        # 填寫任務名稱，必填。
        job_name = "examplejob"
        # 填寫任務ID，必填。
        version = "b4155550-****-4371-****-9c7337348021"
        # 填寫任務輪次，必填。
        runtime_id = 1
        client.create_report(userid, CreateReportRequest(
            CreateReportInfo(
                job_name=job_name,
                version=version,
                runtime_id=runtime_id
            )
        ))
    except Exception as e:
        print(e)

擷取任務遷移報告

說明

調用該方法可以查詢到指定任務的遷移報告，可根據該方法判斷遷移報告是否產生完畢以及擷取產生的遷移報告。

以下範例程式碼用於查詢指定任務遷移報告。

import os

from alibabacloud_hcs_mgw20240626.client import Client
from alibabacloud_hcs_mgw20240626.models import GetReportRequest
from alibabacloud_tea_openapi.models import Config

if __name__ == "__main__":
    try:
        # 填寫主帳號ID。
        userid = "11470***876***55"
        # 這裡以北京地區為例。
        endpoint = "cn-beijing.mgw.aliyuncs.com"
        # 支援 HTTPS、HTTP, 未指定時預設採用 HTTPs。
        config = Config(
            endpoint=endpoint,
            access_key_id=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_ID"),
            access_key_secret=os.environ.get("ALIBABA_CLOUD_ACCESS_KEY_SECRET"),
            protocol="http"
        )
        client = Client(config)

        # 填寫任務ID，必填。
        version = "b4155550-****-4371-****-9c7337348021"
        # 填寫任務輪次，必填。
        runtime_id = 1
        response = client.get_report(userid, GetReportRequest(
            runtime_id=runtime_id,
            version=version
        ))
        print(response.body.get_report_response)
    except Exception as e:
        print(e)