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

本文針對檔案的傳輸情境，介紹如何使用Python SDK V2新增的上傳管理器Uploader模組進行檔案上傳。

注意事項

本文範例程式碼以華東1（杭州）的地區IDcn-hangzhou為例，預設使用外網Endpoint，如果您希望通過與OSS同地區的其他阿里雲產品訪問OSS，請使用內網Endpoint。關於OSS支援的Region與Endpoint的對應關係，請參見OSS地區和訪問網域名稱。
要進行上傳，您必須有oss:PutObject許可權。具體操作，請參見為RAM使用者授予自訂的權限原則。

方法定義

上傳管理器功能簡介

Python SDK V2新增上傳管理器Uploader提供了通用的上傳方法，隱藏了底層方法的實現細節，提供便捷的檔案上傳能力。

上傳管理器Uploader底層利用分區上傳方法，把檔案或者流分成多個較小的分區並發上傳，提升上傳的效能。
上傳管理器Uploader同時提供了斷點續傳的能力，即在上傳過程中，記錄已完成的分區狀態，如果出現網路中斷、程式異常退出等問題導致檔案上傳失敗，甚至重試多次仍無法完成上傳，再次上傳時，可以通過斷點記錄檔案恢複上傳。

上傳管理器Uploader的常用方法如下：

class Uploader:
  ...

def uploader(self, **kwargs) -> Uploader:
  ...

def upload_file(self, request: models.PutObjectRequest, filepath: str, **kwargs: Any) -> UploadResult:
  ...
  
def upload_from(self, request: models.PutObjectRequest, reader: IO[bytes], **kwargs: Any) -> UploadResult:
  ...

請求參數列表

參數名	類型	說明
request	PutObjectRequest	上傳對象的請求參數，和PutObject 方法的請求參數一致，具體請參見PutObjectRequest
reader	IO[bytes]	需要上傳的資料流
filepath	str	本地檔案路徑
**kwargs	Any	(可選)任意參數，類型為字典

返回參數列表

類型	說明
UploadResult	上傳對象的返回參數，具體請參見UploadResult

當您使用client.uploader初始化上傳管理器執行個體時，您可以指定多個配置選項來自訂上傳行為，也可以在每次調用上傳介面時，指定多個配置選項來自訂每次上傳對象的行為。例如指定分區大小如下所示。

設定uploader的配置參數

uploader = client.uploader(part_size=10  * 1024 * 1024)

設定每次上傳請求的配置參數

result = uploader.upload_file(oss.PutObjectRequest(
        bucket="example_bucket",
        key="example_key",
    ),
    filepath="/local/dir/example",
    part_size=10 * 1024 * 1024,
)

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

參數名	類型	說明
part_size	int	指定分區大小，預設值為 6MiB
parallel_num	int	指定上傳任務的並發數，預設值為 3。針對的是單次調用的並發限制，而不是全域的並發限制
leave_parts_on_error	bool	當上傳失敗時，是否保留已上傳的分區，預設不保留
enable_checkpoint	bool	是否開啟斷點上傳功能，預設不開啟說明 enable_checkpoint參數目前僅對upload_file方法有效，upload_from方法暫不支援
checkpoint_dir	str	指定記錄檔案的儲存路徑，例如 /local/dir/, 當enable_checkpoint 為 true時有效

關於檔案上傳管理器方法的完整定義，請參見Uploader。

範例程式碼

您可以通過以下代碼使用上傳管理器上傳本地檔案到儲存空間。

import argparse
import alibabacloud_oss_v2 as oss

# 建立一個命令列參數解析器，並描述指令碼用途：上傳檔案樣本
parser = argparse.ArgumentParser(description="upload file sample")

# 添加命令列參數 --region，表示儲存空間所在的地區，必需參數
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令列參數 --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，表示對象（檔案）在OSS中的鍵名，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)
# 添加命令列參數 --file_path，表示本地待上傳檔案的路徑，必需參數，例如“/Users/yourLocalPath/yourFileName”
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

def main():
    # 解析命令列提供的參數，擷取使用者輸入的值
    args = parser.parse_args()

    # 從環境變數中載入訪問OSS所需的認證資訊，用於身分識別驗證
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置建立設定物件，並設定認證提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # 設定設定物件的地區屬性，根據使用者提供的命令列參數
    cfg.region = args.region

    # 如果提供了自訂endpoint，則更新設定物件中的endpoint屬性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS用戶端，準備與OSS互動
    client = oss.Client(cfg)

    # 建立一個用於上傳檔案的對象
    uploader = client.uploader()

    # 調用方法執行檔案上傳操作
    result = uploader.upload_file(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 指定目標儲存空間
            key=args.key,        # 指定檔案在OSS中的名稱
        ),
        filepath=args.file_path  # 指定本地檔案的位置
    )

    # 列印上傳結果的相關資訊，包括狀態代碼、請求ID、內容MD5等
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.headers.get("Content-MD5")},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
          f' server time: {result.headers.get("x-oss-server-time")},'
          )

# 當此指令碼被直接執行時，調用main函數開始處理邏輯
if __name__ == "__main__":
    main()  # 指令碼進入點，控製程序流程從這裡開始

常見使用情境

使用上傳管理器啟動斷點續傳功能

您可以使用以下代碼設定上傳管理器Uploader的配置參數，啟動斷點續傳功能。

import argparse
import alibabacloud_oss_v2 as oss

# 建立一個命令列參數解析器，並描述指令碼用途：上傳檔案樣本
parser = argparse.ArgumentParser(description="upload file sample")

# 添加命令列參數 --region，表示儲存空間所在的地區，必需參數
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令列參數 --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，表示對象（檔案）在OSS中的鍵名，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)
# 添加命令列參數 --file_path，表示本地待上傳檔案的路徑，必需參數，例如“/Users/yourLocalPath/yourFileName”
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

def main():
    # 解析命令列提供的參數，擷取使用者輸入的值
    args = parser.parse_args()

    # 從環境變數中載入訪問OSS所需的認證資訊，用於身分識別驗證
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置建立設定物件，並設定認證提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # 設定設定物件的地區屬性，根據使用者提供的命令列參數
    cfg.region = args.region

    # 如果提供了自訂endpoint，則更新設定物件中的endpoint屬性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS用戶端，準備與OSS互動
    client = oss.Client(cfg)

    # 建立一個用於上傳檔案的對象，並開啟斷點續傳功能，指定斷點記錄檔案的儲存路徑
    uploader = client.uploader(enable_checkpoint=True, checkpoint_dir="/Users/yourLocalPath/checkpoint/")

    # 調用方法執行檔案上傳操作
    result = uploader.upload_file(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 指定目標儲存空間
            key=args.key,        # 指定檔案在OSS中的名稱
        ),
        filepath=args.file_path  # 指定本地檔案的位置
    )

    # 列印上傳結果的相關資訊，包括狀態代碼、請求ID、內容MD5等
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.headers.get("Content-MD5")},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
          f' server time: {result.headers.get("x-oss-server-time")},'
          )

# 當此指令碼被直接執行時，調用main函數開始處理邏輯
if __name__ == "__main__":
    main()  # 指令碼進入點，控製程序流程從這裡開始

使用上傳管理器上傳本地檔案流

您可以通過以下代碼使用上傳管理器上傳本地檔案流。

import argparse
import alibabacloud_oss_v2 as oss

# 建立一個命令列參數解析器，並描述指令碼用途：從檔案上傳樣本
parser = argparse.ArgumentParser(description="upload from sample")

# 添加命令列參數 --region，表示儲存空間所在的地區，必需參數
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令列參數 --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，表示對象（檔案）在OSS中的鍵名，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)
# 添加命令列參數 --file_path，表示本地待上傳檔案的路徑，必需參數，例如“/Users/yourLocalPath/yourFileName”
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

def main():
    # 解析命令列提供的參數，擷取使用者輸入的值
    args = parser.parse_args()

    # 從環境變數中載入訪問OSS所需的認證資訊，用於身分識別驗證
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置建立設定物件，並設定認證提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # 設定設定物件的地區屬性，根據使用者提供的命令列參數
    cfg.region = args.region

    # 如果提供了自訂endpoint，則更新設定物件中的endpoint屬性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS用戶端，準備與OSS互動
    client = oss.Client(cfg)

    # 建立一個用於上傳檔案的對象
    uploader = client.uploader()

    # 開啟本地檔案以二進位模式讀取
    with open(file=args.file_path, mode='rb') as f:
        # 調用方法執行檔案上傳操作
        result = uploader.upload_from(
            oss.PutObjectRequest(
                bucket=args.bucket,  # 指定目標儲存空間
                key=args.key,        # 指定檔案在OSS中的名稱
            ),
            reader=f  # 傳入檔案讀取器
        )

        # 列印上傳結果的相關資訊，包括狀態代碼、請求ID、內容MD5等
        print(f'status code: {result.status_code},'
              f' request id: {result.request_id},'
              f' content md5: {result.headers.get("Content-MD5")},'
              f' etag: {result.etag},'
              f' hash crc64: {result.hash_crc64},'
              f' version id: {result.version_id},'
              f' server time: {result.headers.get("x-oss-server-time")},'
              )

# 當此指令碼被直接執行時，調用main函數開始處理邏輯
if __name__ == "__main__":
    main()  # 指令碼進入點，控製程序流程從這裡開始

使用上傳管理器設定分區大小和並發數

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

import argparse
import alibabacloud_oss_v2 as oss

# 建立一個命令列參數解析器，並描述指令碼用途：上傳檔案樣本
parser = argparse.ArgumentParser(description="upload file sample")

# 添加命令列參數 --region，表示儲存空間所在的地區，必需參數
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令列參數 --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，表示對象（檔案）在OSS中的鍵名，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)
# 添加命令列參數 --file_path，表示本地待上傳檔案的路徑，必需參數，例如“/Users/yourLocalPath/yourFileName”
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

def main():
    # 解析命令列提供的參數，擷取使用者輸入的值
    args = parser.parse_args()

    # 從環境變數中載入訪問OSS所需的認證資訊，用於身分識別驗證
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置建立設定物件，並設定認證提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    
    # 設定設定物件的地區屬性，根據使用者提供的命令列參數
    cfg.region = args.region

    # 如果提供了自訂endpoint，則更新設定物件中的endpoint屬性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS用戶端，準備與OSS互動
    client = oss.Client(cfg)

    # 建立一個用於上傳檔案的對象，並設定分區大小和並發數
    uploader = client.uploader(
        part_size=100 * 1024,  # 設定分區大小為100KB
        parallel_num=5,        # 設定並發數為5
        leave_parts_on_error=True  # 在發生錯誤時保留已上傳的部分
    )

    # 調用方法執行檔案上傳操作
    result = uploader.upload_file(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 指定目標儲存空間
            key=args.key,        # 指定檔案在OSS中的名稱
        ),
        filepath=args.file_path  # 指定本地檔案的位置
    )

    # 列印上傳結果的相關資訊，包括狀態代碼、請求ID、內容MD5等
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.headers.get("Content-MD5")},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
          f' server time: {result.headers.get("x-oss-server-time")},'
          )

# 當此指令碼被直接執行時，調用main函數開始處理邏輯
if __name__ == "__main__":
    main()  # 指令碼進入點，控製程序流程從這裡開始

使用上傳管理器設定上傳回調

如果您希望在檔案上傳後通知應用伺服器，可參考以下程式碼範例。

import argparse
import base64
import alibabacloud_oss_v2 as oss

# 建立一個命令列參數解析器，並描述指令碼用途：上傳檔案樣本
parser = argparse.ArgumentParser(description="upload file sample")

# 添加命令列參數 --region，表示儲存空間所在的地區，必需參數
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
# 添加命令列參數 --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，表示對象（檔案）在OSS中的鍵名，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)
# 添加命令列參數 --file_path，表示本地待上傳檔案的路徑，必需參數
parser.add_argument('--file_path', help='The path of Upload file.', required=True)

def main():
    # 解析命令列提供的參數，擷取使用者輸入的值
    args = parser.parse_args()

    # 從環境變數中載入訪問OSS所需的認證資訊，用於身分識別驗證
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 使用SDK的預設配置建立設定物件，並設定認證提供者
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider

    # 設定設定物件的地區屬性，根據使用者提供的命令列參數
    cfg.region = args.region

    # 如果提供了自訂endpoint，則更新設定物件中的endpoint屬性
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 使用上述配置初始化OSS用戶端，準備與OSS互動
    client = oss.Client(cfg)

    # 建立一個用於上傳檔案的uploader對象
    uploader = client.uploader()

    # 定義回調地址
    call_back_url = "http://www.example.com/callback"
    # 構造回調參數（callback）：指定回調地址和回調請求體，使用 Base 64 編碼
    callback=base64.b64encode(str('{\"callbackUrl\":\"' + call_back_url + '\",\"callbackBody\":\"bucket=${bucket}&object=${object}&my_var_1=${x:var1}&my_var_2=${x:var2}\"}').encode()).decode()
    # 構造自訂變數（callback-var），使用 Base 64 編碼
    callback_var=base64.b64encode('{\"x:var1\":\"value1\",\"x:var2\":\"value2\"}'.encode()).decode()

    # 調用方法執行檔案上傳操作
    result = uploader.upload_file(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 指定目標儲存空間
            key=args.key,        # 指定檔案在OSS中的名稱
            callback=callback,
            callback_var=callback_var,
        ),
        filepath=args.file_path,  # 指定本地檔案的位置
    )

    # 列印上傳結果的相關資訊，包括狀態代碼、請求ID、內容MD5等
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.headers.get("Content-MD5")},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
          f' server time: {result.headers.get("x-oss-server-time")},'
          )

# 當此指令碼被直接執行時，調用main函數開始處理邏輯
if __name__ == "__main__":
    main()  # 指令碼進入點，控製程序流程從這裡開始

Object Storage Service：檔案上傳管理器（Python SDK V2）