如何通過Python SDK V2上傳檔案 - Object Storage Service

本文介紹如何通過簡單上傳方法將單個檔案快速上傳到OSS，此方法操作簡便，適合快速將檔案上傳到雲端儲存。

注意事項

本文範例程式碼以華東1（杭州）的地區IDcn-hangzhou為例，預設使用外網Endpoint，如果您希望通過與OSS同地區的其他阿里雲產品訪問OSS，請使用內網Endpoint。關於OSS支援的Region與Endpoint的對應關係，請參見OSS地區和訪問網域名稱。

許可權說明

阿里雲帳號預設擁有全部許可權。阿里雲帳號下的RAM使用者或RAM角色預設沒有任何許可權，需要阿里雲帳號或帳號管理員通過RAM Policy或Bucket Policy授予操作許可權。

API	Action	說明
PutObject	`oss:PutObject`	上傳Object。
	`oss:PutObjectTagging`	上傳Object時，如果通過x-oss-tagging指定Object的標籤，則需要此操作的許可權。
	`kms:GenerateDataKey`	上傳Object時，如果Object的中繼資料套件含X-Oss-Server-Side-Encryption: KMS，則需要這兩個操作的許可權。
	`kms:Decrypt`

方法定義

put_object(request: PutObjectRequest, **kwargs) → PutObjectResult

請求參數列表

參數名	類型	說明
request	PutObjectRequest	佈建要求參數，例如設定對象的存取控制方式（Acl）、禁止覆蓋寫（ForbidOverwrite）、自訂中繼資料（Metadata）等，具體請參見PutObjectRequest

傳回值列表

類型	說明
PutObjectResult	傳回值，具體請參見PutObjectResult

關於簡單上傳方法的完整定義，請參見put_object。

上傳本地檔案

上傳檔案（Object）時，如果儲存空間（Bucket）中已存在同名檔案且使用者對該檔案有存取權限，則新添加的檔案將覆蓋原有檔案。

上傳檔案時涉及填寫的公用參數如下：

參數	說明
bucket_name	Bucket名稱。 Bucket名稱的命名規範如下：只能包括小寫字母、數字和短劃線（-）。必須以小寫字母或者數字開頭和結尾。長度必須在3~63字元之間。
object_name	Object完整路徑。Object完整路徑中不能包含Bucket名稱。 Object命名規範如下：使用UTF-8編碼。長度必須在1~1023字元之間。不能以正斜線（/）或者反斜線（\）開頭。

使用put_object_from_file方法直接將本地檔案上傳到目標儲存空間。

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="put object from 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，表示對象的名稱，必需參數
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()

    # 從環境變數中載入憑證資訊，用於身分識別驗證
    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用戶端
    client = oss.Client(cfg)

    # 執行上傳對象的請求，直接從檔案上傳
    # 指定儲存空間名稱、對象名稱和本地檔案路徑
    result = client.put_object_from_file(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 儲存空間名稱
            key=args.key         # 對象名稱
        ),
        args.file_path          # 本地檔案路徑
    )

    # 輸出請求的結果資訊，包括狀態代碼、請求ID、內容MD5、ETag、CRC64校正碼、版本ID和伺服器回應時間
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.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()

使用put_object方法上傳本地檔案，必須使用'rb'模式開啟檔案，以確保上傳的是原始位元組流而非常值內容，避免出現CRC校正失敗。

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="put object from 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，表示對象的名稱，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)

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

    # 從環境變數中載入憑證資訊，用於身分識別驗證
    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用戶端
    client = oss.Client(cfg)

    # 執行上傳對象的請求，直接從本地檔案上傳
    # 指定儲存空間名稱、對象名稱和本地檔案路徑
    with open('your-test-file.md', 'rb') as f:
    result = client.put_object(
        oss.PutObjectRequest(
            bucket=args.bucket,  # 儲存空間名稱
            key=args.key,        # 對象名稱
            body=f.read()        # 讀取檔案內容
        )
    )

    # 輸出請求的結果資訊，包括狀態代碼、請求ID、內容MD5、ETag、CRC64校正碼、版本ID和伺服器回應時間
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.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="put object 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，表示對象的名稱，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)

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

    # 從環境變數中載入憑證資訊，用於身分識別驗證
    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用戶端
    client = oss.Client(cfg)

    # 定義要上傳的字串內容
    text_string = "Hello, OSS!"
    data = text_string.encode('utf-8')  # 將字串編碼為UTF-8位元組串

    # 執行上傳對象的請求，指定儲存空間名稱、對象名稱和資料內容
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=data,
    ))

    # 輸出請求的結果狀態代碼、請求ID、內容MD5、ETag、CRC64校正碼和版本ID，用於檢查請求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 指令碼入口，當檔案被直接運行時調用main函數

上傳位元組數組

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="put object 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，表示對象的名稱，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)

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

    # 從環境變數中載入憑證資訊，用於身分識別驗證
    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用戶端
    client = oss.Client(cfg)

    # 定義要上傳的資料內容
    data = b'hello world'

    # 執行上傳對象的請求，指定儲存空間名稱、對象名稱和資料內容
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=data,
    ))

    # 輸出請求的結果狀態代碼、請求ID、內容MD5、ETag、CRC64校正碼和版本ID，用於檢查請求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 指令碼入口，當檔案被直接運行時調用main函數

上傳網路流

您可以使用以下代碼將網路流上傳至目標儲存空間。

import argparse
import requests
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="put object 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，表示對象的名稱，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)

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

    # 從環境變數中載入憑證資訊，用於身分識別驗證
    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用戶端
    client = oss.Client(cfg)

    # 發送HTTP GET請求，擷取響應內容
    response = requests.get('http://www.aliyun.com')

    # 執行上傳對象的請求，指定儲存空間名稱、對象名稱和資料內容
    result = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=response.content,
    ))

    # 輸出請求的結果狀態代碼、請求ID、內容MD5、ETag、CRC64校正碼和版本ID，用於檢查請求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 指令碼入口，當檔案被直接運行時調用main函數

上傳檔案並設定回調

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

import base64
import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="put object sample")

# 添加必要的參數
parser.add_argument('--region', help='The region in which the bucket is located.', required=True)
parser.add_argument('--bucket', help='The name of the bucket.', required=True)
parser.add_argument('--endpoint', help='The domain names that other services can use to access OSS')
parser.add_argument('--key', help='The name of the object.', required=True)
parser.add_argument('--call_back_url', help='Callback server address.', required=True)


def main():

    args = parser.parse_args()

    # 從環境變數中載入憑證資訊
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    # 配置 SDK 用戶端
    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    # 建立 OSS 用戶端
    client = oss.Client(cfg)

    # 要上傳的內容（字串）
    data = 'hello world'

    # 構造回調參數（callback）：指定回調地址和回調請求體，使用 Base 64 編碼
    callback=base64.b64encode(str('{\"callbackUrl\":\"' + args.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 = client.put_object(oss.PutObjectRequest(
        bucket=args.bucket,
        key=args.key,
        body=data,
        callback=callback,
        callback_var=callback_var,
    ))
    # 列印返回結果（包括狀態代碼、請求 ID 等）
    print(vars(result))


if __name__ == "__main__":
    main()

上傳檔案顯示進度條

以下範例程式碼展示了在上傳本地檔案時，使用進度條查看上傳進度。

import argparse
import alibabacloud_oss_v2 as oss

# 建立命令列參數解析器
parser = argparse.ArgumentParser(description="put object 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，表示對象的名稱，必需參數
parser.add_argument('--key', help='The name of the object.', required=True)

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

    # 從環境變數中載入憑證資訊，用於身分識別驗證
    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用戶端
    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='')

    # 執行上傳對象的請求，指定儲存空間名稱、對象名稱和資料內容
    result = client.put_object_from_file(oss.PutObjectRequest(
            bucket=args.bucket,
            key=args.key,
            progress_fn=_progress_fn,
        ),
        "/local/dir/example", # 指定本地檔案路徑
    )

    # 輸出請求的結果狀態代碼、請求ID、內容MD5、ETag、CRC64校正碼和版本ID，用於檢查請求是否成功
    print(f'status code: {result.status_code},'
          f' request id: {result.request_id},'
          f' content md5: {result.content_md5},'
          f' etag: {result.etag},'
          f' hash crc64: {result.hash_crc64},'
          f' version id: {result.version_id},'
    )

if __name__ == "__main__":
    main()  # 指令碼入口，當檔案被直接運行時調用main函數

Object Storage Service：簡單上傳（Python SDK V2）