全部產品
Search

搜尋本產品

    Object Storage Service：Python拷貝檔案

    文件中心

    Object Storage Service：Python拷貝檔案

    更新時間：Oct 25, 2024

    本文介紹如何將源Bucket中的檔案（Object）複製到同一地區下相同或不同目標Bucket中。

    注意事項

    • 本文以華東1（杭州）外網Endpoint為例。如果您希望通過與OSS同地區的其他阿里雲產品訪問OSS，請使用內網Endpoint。關於OSS支援的Region與Endpoint的對應關係，請參見OSS訪問網域名稱、資料中心、開放連接埠。

    • 本文以從環境變數讀取存取憑證為例。如何配置訪問憑證，請參見配置訪問憑證。

    • 本文以OSS網域名稱建立OSSClient為例。如果您希望通過自訂網域名、STS等方式建立OSSClient，請參見初始化。

    • 拷貝檔案時，您必須擁有源檔案的讀許可權及目標Bucket的讀寫權限。

    • 拷貝檔案時，您需要確保源Bucket和目標Bucket均未設定合規保留原則，否則報錯The object you specified is immutable.。

    • 不支援跨地區拷貝。例如不能將華東1（杭州）地區儲存空間中的檔案拷貝到華北1（青島）地區。

    拷貝小檔案

    對於小於1 GB的檔案，您可以使用簡單拷貝。以下代碼用於通過簡單拷貝將源儲存空間srcexamplebucket中的srcexampleobject.txt檔案拷貝到目標儲存空間destexamplebucket中的destexampleobject.txt檔案。

    # -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider

# 從環境變數中擷取訪問憑證。運行本程式碼範例之前，請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# 填寫源Bucket名稱，例如srcexamplebucket。
src_bucket_name = 'srcexamplebucket'
# 填寫與源Bucket處於同一地區的目標Bucket名稱，例如destexamplebucket。
# 當在同一個Bucket內拷貝檔案時，請確保源Bucket名稱和目標Bucket名稱相同。
dest_bucket_name = 'destexamplebucket'

# 填寫Bucket所在地區對應的Endpoint。以華東1（杭州）為例，Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# 填寫Endpoint對應的Region資訊，例如cn-hangzhou。注意，v4簽名下，必須填寫該參數
region = "cn-hangzhou"
bucket = oss2.Bucket(auth, endpoint, dest_bucket_name, region=region)


# 填寫不包含Bucket名稱在內源Object的完整路徑，例如srcexampleobject.txt。
src_object_name = 'srcexampleobject.txt'
# 填寫不包含Bucket名稱在內目標Object的完整路徑，例如destexampleobject.txt。
dest_object_name = 'destexampleobject.txt'

# headers = dict()
# 指定CopyObject操作時是否覆蓋同名目標Object。此處設定為true，表示禁止覆蓋同名Object。
# headers['x-oss-forbid-overwrite'] = 'true'
# 指定拷貝的源地址。
# headers[OSS_COPY_OBJECT_SOURCE] = '/example-bucket-by-util/recode-test.txt'
# 如果源Object的ETag值和您提供的ETag相等，則執行拷貝操作，並返回200 OK。
# headers['x-oss-copy-source-if-match'] = '5B3C1A2E053D763E1B002CC607C5****'
# 如果源Object的ETag值和您提供的ETag不相等，則執行拷貝操作，並返回200 OK。
# headers['x-oss-copy-source-if-none-match'] = '5B3C1A2E053D763E1B002CC607C5****'
# 如果指定的時間等於或者晚於檔案實際修改時間，則正常拷貝檔案，並返回200 OK。
# headers['x-oss-copy-source-if-unmodified-since'] = '2021-12-09T07:01:56.000Z'
# 如果源Object在指定時間後被修改過，則執行拷貝操作。
# headers['x-oss-copy-source-if-modified-since'] = '2021-12-09T07:01:56.000Z'
# 指定設定目標Object中繼資料的方式。此處設定為COPY，表示複製源Object的中繼資料到目標Object。
# headers[OSS_METADATA_DIRECTIVE] = 'COPY'
# 指定OSS建立目標Object時使用的伺服器端密碼編譯演算法。
# headers[OSS_SERVER_SIDE_ENCRYPTION] = 'KMS'
# 表示KMS託管的使用者主要金鑰，該參數僅在x-oss-server-side-encryption為KMS時有效。
# headers['x-oss-server-side-encryption-key-id'] = '9468da86-3509-4f8d-a61e-6eab1eac****'
# 指定OSS建立目標Object時的存取權限。此處設定為OBJECT_ACL_PRIVATE，表示只有Object的擁有者和授權使用者有該Object的讀寫權限，其他使用者沒有許可權操作該Object。
# headers[OSS_OBJECT_ACL] = oss2.OBJECT_ACL_PRIVATE
# 指定Object的儲存類型。此處設定為BUCKET_STORAGE_CLASS_STANDARD，表示標準儲存類型。
# headers['x-oss-storage-class'] = oss2.BUCKET_STORAGE_CLASS_STANDARD
# 指定Object的對象標籤，可同時設定多個標籤。
# headers[OSS_OBJECT_TAGGING] = 'k1=v1&k2=v2&k3=v3'
# 指定設定目標Object對象標籤的方式。此處設定為COPY，表示複製源Object的對象標籤到目標Object。
# headers[OSS_OBJECT_TAGGING_COPY_DIRECTIVE] = 'COPY'
# result = bucket.copy_object(src_bucket_name, src_object_name, dest_object_name, headers=headers)

# 將源Bucket中的某個Object拷貝到目標Bucket。
result = bucket.copy_object(src_bucket_name, src_object_name, dest_object_name)

# 查看返回結果的狀態。如果傳回值為200，表示執行成功。
print('result.status:', result.status)

    拷貝大檔案

    對於大於1 GB的檔案，需要使用分區拷貝（UploadPartCopy）。分區拷貝分為三步：

    1. 通過bucket.init_multipart_upload初始化分區拷貝任務。

    2. 通過bucket.upload_part_copy進行分區拷貝。除最後一個分區外，其它分區都要大於100 KB。

    3. 通過bucket.complete_multipart_upload提交分區拷貝任務。

    以下代碼用於通過分區拷貝將源儲存空間srcexamplebucket中的srcexampleobject.txt檔案拷貝到目標儲存空間destexamplebucket中的destexampleobject.txt檔案。

    # -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
from oss2.models import PartInfo
from oss2 import determine_part_size

# 從環境變數中擷取訪問憑證。運行本程式碼範例之前，請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# 填寫源Bucket名稱，例如srcexamplebucket。
src_bucket_name = 'srcexamplebucket'
# 填寫與源Bucket處於同一地區的目標Bucket名稱，例如destexamplebucket。
# 當在同一Bucket內拷貝檔案時，請確保源Bucket名稱和目標Bucket名稱相同。
dest_bucket_name = 'destexamplebucket'

# 填寫Bucket所在地區對應的Endpoint。以華東1（杭州）為例，Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# 填寫Endpoint對應的Region資訊，例如cn-hangzhou。注意，v4簽名下，必須填寫該參數
region = "cn-hangzhou"

# 當在同一Bucket內拷貝檔案時，請注釋掉該行代碼，並將後面的src_bucket改為bucket即可。
bucket = oss2.Bucket(auth, endpoint, dest_bucket_name, region=region)

src_bucket = oss2.Bucket(auth, endpoint, src_bucket_name, region=region)

# 填寫不包含Bucket名稱在內源Object的完整路徑，例如srcexampleobject.txt。
src_object_name = 'srcexampleobject.txt'
# 填寫不包含Bucket名稱在內目標Object的完整路徑，例如destexampleobject.txt。
dest_object_name = 'destexampleobject.txt'
# 擷取源檔案的大小。當在同一個Bucket內拷貝檔案時，請將src_bucket改為bucket。
head_info = src_bucket.head_object(src_object_name)
total_size = head_info.content_length
print('src object size:', total_size)

# determine_part_size方法用來確定分區大小。
part_size = determine_part_size(total_size, preferred_size=100 * 1024)
print('part_size:', part_size)

# 初始化分區。
upload_id = bucket.init_multipart_upload(dest_object_name).upload_id
parts = []

# 逐個上傳分區。
part_number = 1
offset = 0
while offset < total_size:
    num_to_upload = min(part_size, total_size - offset)
    end = offset + num_to_upload - 1
    # headers = dict()
    # 指定拷貝的源地址。 
    # headers[OSS_COPY_OBJECT_SOURCE] = '/example-bucket-by-util/recode-test.txt'
    # 指定源Object的拷貝範圍。例如設定bytes=0~1023，表示拷貝1~1024位元組的內容。
    # headers[OSS_COPY_OBJECT_SOURCE_RANGE] = 'bytes=0~1023'
    # 如果源Object的ETag值和您提供的ETag相等，則執行拷貝操作，並返回200 OK。
    # headers['x-oss-copy-source-if-match'] = '5B3C1A2E053D763E1B002CC6****'
    # 如果源Object的ETag值和您提供的ETag不相等，則執行拷貝操作，並返回200 OK。
    # headers['x-oss-copy-source-if-none-match'] = '5B3C1A2E053D763E1B002CC6****'
    # 如果指定的時間等於或者晚於檔案實際修改時間，則正常拷貝檔案，並返回200 OK。
    # headers['x-oss-copy-source-if-unmodified-since'] = '2021-12-09T07:01:56.000Z'
    # 如果指定的時間早於檔案實際修改時間，則正常拷貝檔案，並返回200 OK。
    # headers['x-oss-copy-source-if-modified-since'] = '2021-12-09T07:01:56.000Z'
    # result = bucket.upload_part_copy(src_bucket_name, src_object_name, (offset, end), dest_object_name, upload_id, part_number, headers=headers)
    
    result = bucket.upload_part_copy(src_bucket_name, src_object_name, (offset, end), dest_object_name, upload_id, part_number)
    # 儲存part資訊。
    parts.append(PartInfo(part_number, result.etag))

    offset += num_to_upload
    part_number += 1

# 完成分區拷貝。
result = bucket.complete_multipart_upload(dest_object_name, upload_id, parts)
# 查看拷貝返回狀態。
print('result :', result.status)
# 擷取檔案中繼資料。
head_info = bucket.head_object(dest_object_name)
# 查看目標Object大小。
dest_object_size = head_info.content_length
print('dest object size:', dest_object_size)
# 對比源Object和目標Object的大小。
assert dest_object_size == total_size

    相關文檔

    • 拷貝小檔案

      • 關於拷貝小檔案的完整範例程式碼，請參見GitHub樣本。

      • 關於拷貝小檔案的API介面說明，請參見CopyObject。

    • 拷貝大檔案

      關於拷貝大檔案的API介面說明，請參見UploadPartCopy。