全部產品
Search
文件中心

Object Storage Service:Python上傳檔案

更新時間:Jun 19, 2024

本文介紹如何在受版本控制的儲存空間(Bucket)中使用簡單上傳、追加上傳或者分區上傳的方式來上傳檔案(Object)。

注意事項

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

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

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

  • 要上傳檔案,您必須有oss:PutObject許可權。具體操作,請參見為RAM使用者授權自訂的權限原則

簡單上傳

在已開啟版本控制的Bucket中,OSS會為新添加的Object自動產生唯一的版本ID,並在響應header中通過x-oss-version-id形式返回。在暫停了版本控制的Bucket中,新添加的Object的版本ID為“null”,上傳同名Object,後一次會覆蓋前一次上傳的檔案內容。OSS保證同一個Object只會有一個版本ID為“null”。

以下代碼用於簡單上傳:

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
# 從環境變數中擷取訪問憑證。運行本程式碼範例之前,請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
# 填寫Bucket所在地區對應的Endpoint。以華東1(杭州)為例,Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
# yourBucketName填寫儲存空間名稱。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'yourBucketName')

# 上傳檔案。
result = bucket.put_object('yourObjectName', 'content of object')
# HTTP返回碼。
print('http response code: {0}'.format(result.status))
# 查看本次上傳Object的版本ID。
print('put object version:', result.versionid)

追加上傳

在受版本控制的Bucket中,僅支援對於目前的版本為Appendable類型的Object執行追加(AppendObject)操作,不支援對於歷史版本為Appendable類型的Object執行AppendObject操作。

說明
  • 對目前的版本為Appendable類型的Object執行AppendObject操作時,OSS不會為該Appendable類型的Object產生歷史版本。

  • 對目前的版本為Appendable類型的Object執行PutObject或DeleteObject操作時,OSS會將該Appendable類型的Object保留為歷史版本,且該Object不允許繼續追加。

  • 不支援對目前的版本為非Appendable類型的Object(包括Normal Object、Delete Marker等)執行AppendObject操作。

以下代碼用於追加上傳:

# -*- coding: utf-8 -*-
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
# 從環境變數中擷取訪問憑證。運行本程式碼範例之前,請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
# 填寫Bucket所在地區對應的Endpoint。以華東1(杭州)為例,Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
# yourBucketName填寫儲存空間名稱。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'yourBucketName')

# 設定首次上傳的追加位置(Position參數)為0。
result = bucket.append_object('yourObjectName', 0, 'content of first append')
# 查看本次執行追加檔案的Object的版本ID。
print('append object versionid:', result.versionid)
# 如果不是首次上傳,可以通過bucket.head_object方法或上次追加傳回值的next_position屬性,得到追加位置。
bucket.append_object('yourObjectName', result.next_position, 'content of second append')

分區上傳

在受版本控制的Bucket中,調用CompleteMultipartUpload介面來完成整個檔案的分區上傳,OSS會為整個檔案產生唯一的版本ID,並在響應header中以x-oss-version-id的形式返回。

以下代碼用於分區上傳:

# -*- coding: utf-8 -*-
import os
import oss2
from oss2.credentials import EnvironmentVariableCredentialsProvider
from oss2 import SizedFileAdapter, determine_part_size
from oss2.models import PartInfo
# 從環境變數中擷取訪問憑證。運行本程式碼範例之前,請確保已設定環境變數OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())
# 填寫Bucket所在地區對應的Endpoint。以華東1(杭州)為例,Endpoint填寫為https://oss-cn-hangzhou.aliyuncs.com。
# yourBucketName填寫儲存空間名稱。
bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'yourBucketName')

key = 'yourObjectName'
filename = 'yourLocalFile'

total_size = os.path.getsize(filename)
# determine_part_size方法用來確定分區大小。
part_size = determine_part_size(total_size, preferred_size=100 * 1024)

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

# 逐個上傳分區。
with open(filename, 'rb') as fileobj:
    part_number = 1
    offset = 0
    while offset < total_size:
        num_to_upload = min(part_size, total_size - offset)
    # SizedFileAdapter(fileobj, size)方法會產生一個新的檔案對象,重新計算起始追加位置。
        result = bucket.upload_part(key, upload_id, part_number,
                                    SizedFileAdapter(fileobj, num_to_upload))
        parts.append(PartInfo(part_number, result.etag))

        offset += num_to_upload
        part_number += 1

# 完成分區上傳。
result = bucket.complete_multipart_upload(key, upload_id, parts)
# 查看response中攜帶的上傳檔案的versionid
print('result.versionid:', result.versionid)


# 驗證分區上傳。
with open(filename, 'rb') as fileobj:
    assert bucket.get_object(key).read() == fileobj.read()

相關文檔