如何使用OSS Python SDK V2進行檔案拷貝 - Object Storage Service

本文針對大檔案的傳輸情境，介紹如何使用Python SDK V2新增的Copier模組進行檔案拷貝。

注意事項

本文範例程式碼以華東1（杭州）的地區IDcn-hangzhou為例，預設使用外網Endpoint，如果您希望通過與OSS同地區的其他阿里雲產品訪問OSS，請使用內網Endpoint。關於OSS支援的Region與Endpoint的對應關係，請參見OSS地區和訪問網域名稱。
要進行拷貝檔案，您必須擁有源檔案的讀許可權及目標Bucket的讀寫權限。
不支援跨地區拷貝。例如不能將華東1（杭州）地區儲存空間中的檔案拷貝到華北1（青島）地區。
拷貝檔案時，您需要確保源Bucket和目標Bucket均未設定合規保留原則，否則報錯The object you specified is immutable.。

方法定義

拷貝管理器介紹

當需要將對象從儲存空間複製到另外一個儲存空間，或者修改對象的屬性時，您可以通過拷貝介面或者分區拷貝介面來完成這個操作。這兩個介面有其適用的情境，例如：

拷貝介面（CopyObject）只適合拷貝 5GiB 以下的對象；
分區拷貝介面（UploadPartCopy）支援拷貝大於5GiB 的對象，但不支援中繼資料指令（x-oss-metadata-directive）和標籤指令（x-oss-tagging-directive）參數，拷貝時需要主動設定需要複製的中繼資料和標籤。

Python SDK V2新增拷貝管理器Copier提供了通用的拷貝介面，隱藏了介面的差異和實現細節，可根據拷貝的請求參數自動選擇合適的介面複製對象。Copier的常用方法定義如下：

class CopyError(exceptions.BaseError):
  ...

def copier(self, **kwargs) -> Copier:
  ...

def copy(self, request: models.CopyObjectRequest, **kwargs: Any) -> CopyResult:
  ...

請求參數列表

參數名	類型	說明
request	CopyObjectRequest	設定具體介面的請求參數，具體請參見CopyObjectRequest
**kwargs	Any	(可選)任意參數，類型為字典

其中，CopyObjectRequest的常用參數列舉如下：

參數名	類型	說明
bucket	str	指定目標儲存空間名稱
key	str	指定目標對象名稱
source_bucket	str	指定源儲存空間名稱
source_key	str	指定來源物件名稱
forbid_overwrite	str	指定CopyObject操作時是否覆蓋同名目標Object
tagging	str	指定Object的對象標籤，可同時設定多個標籤，例如TagA=A&TagB=B。
tagging_directive	str	指定如何設定目標Object的對象標籤。取值如下： Copy（預設值）：複製源Object的對象標籤到目標 Object。 Replace：忽略源Object的對象標籤，直接採用請求中指定的對象標籤。

當您使用client.copier初始化拷貝管理器執行個體時，您可以指定多個配置選項來自訂對象的拷貝行為。也可以在每次調用拷貝介面時，指定多個配置選項來自訂每次拷貝對象的行為。

設定Copier的配置參數

copier = client.copier(
    part_size=100 * 1024 * 1024,
)

設定每次拷貝請求的配置參數

result = copier.copy(oss.CopyObjectRequest(
        bucket="example_bucket",
        key="example_key",
        source_bucket="example_source_bucket",
        source_key="example_source_key",
    ),
    part_size=100 * 1024 * 1024,
)

常用的配置選項說明列舉如下：

參數名	類型	說明
part_size	int	指定分區大小，預設值為 64MiB
parallel_num	int	指定拷貝任務的並發數，預設值為 3。針對的是單次調用的並發限制，而不是全域的並發限制
multipart_copy_threshold	int64	使用分區拷貝的閾值，預設值為 200MiB
leave_parts_on_error	bool	當拷貝失敗時，是否保留已拷貝的分區，預設不保留
disable_shallow_copy	bool	不使用淺拷貝行為，預設使用

範例程式碼

您可以使用以下代碼將對象從源儲存空間拷貝到目標儲存空間。

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="copier sample")

# 添加命令列參數：region（必填），指定Bucket所在的地區
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令列參數：bucket（必填），指定目標Bucket名稱
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令列參數：endpoint（可選），指定OSS的訪問網域名稱
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 添加命令列參數：key（必填），指定目標對象的名稱
parser.add_argument('--key', help='The name of the object.', required=True)

# 添加命令列參數：source_key（必填），指定來源物件的名稱
parser.add_argument('--source_key', help='The name of the source address for object.', required=True)

# 添加命令列參數：source_bucket（必填），指定源Bucket的名稱
parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)


def main():
    # 解析命令列參數
    args = parser.parse_args()

    # 從環境變數中載入憑證資訊
    # 使用EnvironmentVariableCredentialsProvider從環境變數中讀取Access Key ID和Access Key Secret
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider  # 設定憑證提供者
    cfg.region = args.region  # 設定Bucket所在的地區
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint  # 如果提供了endpoint，則設定自訂訪問網域名稱

    # 建立OSS用戶端執行個體
    client = oss.Client(cfg)

    # 建立Copier執行個體並執行對象複製操作
    copier = client.copier()

    # 執行對象複製操作
    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,          # 目標Bucket名稱
            key=args.key,                # 目標對象名稱
            source_bucket=args.source_bucket,  # 源Bucket名稱
            source_key=args.source_key   # 來源物件名稱
        )
    )

    # 列印複製結果
    # 使用vars(result)將結果對象轉換為字典格式並列印
    print(vars(result))


if __name__ == "__main__":
    main()

常見使用情境

使用拷貝管理器設定分區大小和並發數

您可以使用以下代碼配置拷貝管理器的參數，設定分區大小和並發數。

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="copier sample")

# 添加命令列參數：region（必填），指定Bucket所在的地區
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令列參數：bucket（必填），指定目標Bucket名稱
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令列參數：endpoint（可選），指定OSS的訪問網域名稱
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 添加命令列參數：key（必填），指定目標對象的名稱
parser.add_argument('--key', help='The name of the object.', required=True)

# 添加命令列參數：source_key（必填），指定來源物件的名稱
parser.add_argument('--source_key', help='The name of the source address for object.', required=True)

# 添加命令列參數：source_bucket（必填），指定源Bucket的名稱
parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)


def main():
    # 解析命令列參數
    args = parser.parse_args()

    # 從環境變數中載入憑證資訊
    # 使用EnvironmentVariableCredentialsProvider從環境變數中讀取Access Key ID和Access Key Secret
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider  # 設定憑證提供者
    cfg.region = args.region  # 設定Bucket所在的地區
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint  # 如果提供了endpoint，則設定自訂訪問網域名稱

    # 建立OSS用戶端執行個體
    client = oss.Client(cfg)

    # 建立Copier執行個體並執行對象複製操作
    copier = client.copier()

    # 執行對象複製操作
    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,          # 目標Bucket名稱
            key=args.key,                # 目標對象名稱
            source_bucket=args.source_bucket,  # 源Bucket名稱
            source_key=args.source_key   # 來源物件名稱
        ),
        part_size= 1 * 1024 * 1024,          # 分區大小，單位為位元組（此處設定為1MiB）
        parallel_num=5,                 # 並發數，控制同時拷貝的分區數量
        leave_parts_on_error=True,      # 如果複製失敗，保留已拷貝的分區
    )

    # 列印複製結果
    # 使用vars(result)將結果對象轉換為字典格式並列印
    print(vars(result))


if __name__ == "__main__":
    main()

使用拷貝管理器顯示進度條

以下範例程式碼展示了在使用拷貝管理器拷貝檔案時，使用進度條查看進度。

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="copier sample")

# 添加命令列參數：region（必填），指定Bucket所在的地區
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)

# 添加命令列參數：bucket（必填），指定目標Bucket名稱
parser.add_argument('--bucket', help='The name of the bucket.', required=True)

# 添加命令列參數：endpoint（可選），指定OSS的訪問網域名稱
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')

# 添加命令列參數：key（必填），指定目標對象的名稱
parser.add_argument('--key', help='The name of the object.', required=True)

# 添加命令列參數：source_key（必填），指定來源物件的名稱
parser.add_argument('--source_key', help='The name of the source address for object.', required=True)

# 添加命令列參數：source_bucket（必填），指定源Bucket的名稱
parser.add_argument('--source_bucket', help='The name of the source address for bucket.', required=True)


def main():
    # 解析命令列參數
    args = parser.parse_args()

    # 從環境變數中載入憑證資訊
    # 使用EnvironmentVariableCredentialsProvider從環境變數中讀取Access Key ID和Access Key Secret
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider  # 設定憑證提供者
    cfg.region = args.region  # 設定Bucket所在的地區
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint  # 如果提供了endpoint，則設定自訂訪問網域名稱

    # 建立OSS用戶端執行個體
    client = oss.Client(cfg)

    # 定義一個字典變數 progress_state 用於儲存拷貝進度狀態，初始值為 0
    progress_state = {'saved': 0}
    def _progress_fn(n, written, total):
        # 使用字典儲存累計寫入的位元組數，避免使用 global 變數
        progress_state['saved'] += n

        # 計算當前拷貝百分比，將已寫入位元組數與總位元組數進行除法運算後取整
        rate = int(100 * (float(written) / float(total)))

        # 列印當前拷貝進度，\r 表示回到行首，實現命令列中即時重新整理效果
        # end='' 表示不換行，使下一次列印覆蓋當前行
        print(f'\r拷貝進度：{rate}% ', end='')

    # 建立Copier執行個體並執行對象複製操作
    copier = client.copier()

    # 執行對象複製操作
    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,          # 目標Bucket名稱
            key=args.key,                # 目標對象名稱
            source_bucket=args.source_bucket,  # 源Bucket名稱
            source_key=args.source_key,   # 來源物件名稱
            progress_fn=_progress_fn,   # 設定進度回呼函數
        )
    )

    # 列印複製結果
    # 使用vars(result)將結果對象轉換為字典格式並列印
    print(vars(result))


if __name__ == "__main__":
    main()

Object Storage Service：檔案拷貝管理器（Python SDK V2）