追加上傳是指在已上傳的追加類型檔案（Appendable Object）末尾直接追加內容。本文介紹如何使用OSS Python SDK進行追加上傳。
注意事項
本文範例程式碼以華東1（杭州）的地區ID
cn-hangzhou為例，預設使用外網Endpoint，如果您希望通過與OSS同地區的其他阿里雲產品訪問OSS，請使用內網Endpoint。關於OSS支援的Region與Endpoint的對應關係，請參見OSS地區和訪問網域名稱。當檔案不存在時，調用追加上傳方法會建立一個追加類型檔案。
當檔案已存在時：
如果檔案為追加類型檔案，且設定的追加位置和檔案當前長度相等，則直接在該檔案末尾追加內容。
如果檔案為追加類型檔案，但是設定的追加位置和檔案當前長度不相等，則拋出PositionNotEqualToLength異常。
如果檔案為非追加類型檔案，例如通過簡單上傳的檔案類型為Normal的檔案，則拋出ObjectNotAppendable異常。
許可權說明
阿里雲帳號預設擁有全部許可權。阿里雲帳號下的RAM使用者或RAM角色預設沒有任何許可權，需要阿里雲帳號或帳號管理員通過RAM Policy或Bucket Policy授予操作許可權。
API | Action | 說明 |
AppendObject |
| 以追加寫的方式上傳檔案（Object）。 |
| 以追加寫的方式上傳檔案（Object）時，如果通過x-oss-tagging指定Object的標籤，則需要此操作的許可權。 |
方法定義
針對檔案追加上傳的情境，Python SDK V2新增了AppendFile方法以模仿檔案的讀寫行為，用於操作儲存空間裡的對象，以下列舉了AppendFile與AppendObject方法的具體說明：
方法名 | 說明 |
AppendFile | 與AppendObject方法能力一致 最佳化了重傳時失敗後容錯處理 |
AppendObject | 追加上傳, 最終檔案最大支援5GiB 支援CRC64資料校正（預設啟用） 支援進度條 |
進階版追加上傳API：AppendFile
調用AppendFile方法以追加寫的方式上傳資料。如果對象不存在，則建立追加類型的對象。如果對象存在，並且不為追加類型的對象，則返回錯誤。
AppendFile方法定義如下。
append_file(bucket: str, key: str, request_payer: str | None = None, create_parameter: AppendObjectRequest | None = None, **kwargs) → AppendOnlyFile請求參數列表
參數名 | 類型 | 說明 |
bucket | str | 設定儲存空間名 |
key | str | 設定對象名 |
RequestPayer | str | 啟用了要求者付費模式時，需要設定為'requester' |
CreateParameter | AppendObjectRequest | 用於首次上傳時，設定對象的元資訊，包括ContentType，Metadata，許可權，儲存類型等，具體請參見AppendObjectRequest |
傳回值列表
類型 | 說明 |
AppendOnlyFile | 追加檔案的執行個體，具體請參見AppendOnlyFile |
其中，AppendOnlyFile類包含的方法說明如下：
方法名 | 說明 |
Close() | 關閉檔案控制代碼，釋放資源 |
write(b) | 將位元組資料寫入到檔案中，返回寫入的位元組數 |
write_from(b: str | bytes | Iterable[bytes] | IO[str] | IO[bytes]) | 將任意資料寫入到檔案中，返回寫入的位元組數 |
關於AppendFile方法的完整定義，請參見append_file。
基礎版追加上傳API：AppendObject
append_object(request: AppendObjectRequest, **kwargs) → AppendObjectResult請求參數列表
參數名 | 類型 | 說明 |
request | AppendObjectRequest | 佈建要求參數，具體請參見AppendObjectRequest |
傳回值列表
類型 | 說明 |
AppendObjectResult | 傳回值，具體請參見AppendObjectResult |
關於AppendObject方法的完整定義，請參見append_object。
範例程式碼
（推薦）使用AppendFile追加上傳
import argparse
import alibabacloud_oss_v2 as oss
# 建立命令列參數解析器，並描述指令碼用途：樣本展示如何向OSS對象追加資料
parser = argparse.ArgumentParser(description="append 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)
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)
# 定義要追加的資料
data1 = b'hello'
data2 = b' world. '
# 第一次追加資料
with client.append_file(bucket=args.bucket, key=args.key) as f:
append_f = f
f.write(data1)
# 列印第一次追加後的檔案狀態
print(f'closed: {append_f.closed},'
f' name: {append_f.name}'
)
# 第二次追加資料
with client.append_file(bucket=args.bucket, key=args.key) as f:
append_f = f
f.write(data2)
# 列印第二次追加後的檔案狀態
print(f'closed: {append_f.closed},'
f' name: {append_f.name}'
)
# 擷取追加後的對象內容
result = client.get_object(oss.GetObjectRequest(
bucket=args.bucket,
key=args.key,
))
# 列印擷取對象的結果
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' content: {result.body.content.decode("utf-8")}'
)
# 當此指令碼被直接執行時，調用main函數開始處理邏輯
if __name__ == "__main__":
main() # 指令碼進入點，控製程序流程從這裡開始使用AppendObject追加上傳
import argparse
import alibabacloud_oss_v2 as oss
# 建立一個命令列參數解析器
parser = argparse.ArgumentParser(description="append object sample")
# 添加命令列參數
# --region: 指定OSS儲存空間所在的地區
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)
def main():
# 解析命令列輸入的參數
args = parser.parse_args()
# 從環境變數中載入OSS所需的認證資訊
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK提供的預設配置建立設定物件
cfg = oss.config.load_default()
# 設定認證資訊提供者為之前建立的對象
cfg.credentials_provider = credentials_provider
# 根據使用者輸入設定OSS用戶端使用的地區
cfg.region = args.region
# 如果使用者提供了自訂的endpoint，則更新配置
if args.endpoint is not None:
cfg.endpoint = args.endpoint
# 使用上述配置建立OSS用戶端執行個體
client = oss.Client(cfg)
# 定義要追加的資料
data1 = b'hello'
data2 = b' world'
# 第一次追加資料
result = client.append_object(oss.AppendObjectRequest(
bucket=args.bucket, # 指定目標儲存空間
key=args.key, # 指定對象的鍵名
position=0, # 追加的起始位置，初始為0
body=data1, # 要追加的資料
))
# 列印第一次追加的結果
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' version id: {result.version_id},'
f' hash crc64: {result.hash_crc64},'
f' next position: {result.next_position},'
)
# 第二次追加資料
result = client.append_object(oss.AppendObjectRequest(
bucket=args.bucket, # 指定目標儲存空間
key=args.key, # 指定對象的鍵名
position=result.next_position, # 從上一次追加的下一個位置開始
body=data2, # 要追加的資料
))
# 列印第二次追加的結果
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' version id: {result.version_id},'
f' hash crc64: {result.hash_crc64},'
f' next position: {result.next_position},'
)
# 當此指令碼被直接運行時，調用main函數
if __name__ == "__main__":
main()常見使用情境
追加上傳顯示進度條
import argparse
import alibabacloud_oss_v2 as oss
# 建立一個命令列參數解析器
parser = argparse.ArgumentParser(description="append object sample")
# 添加命令列參數
# --region: 指定OSS儲存空間所在的地區
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)
def main():
# 解析命令列輸入的參數
args = parser.parse_args()
# 從環境變數中載入OSS所需的認證資訊
credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()
# 使用SDK提供的預設配置建立設定物件
cfg = oss.config.load_default()
# 設定認證資訊提供者為之前建立的對象
cfg.credentials_provider = credentials_provider
# 根據使用者輸入設定OSS用戶端使用的地區
cfg.region = args.region
# 如果使用者提供了自訂的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='')
# 定義要追加的資料
data1 = b'hello'
data2 = b' world'
# 第一次追加資料
result = client.append_object(oss.AppendObjectRequest(
bucket=args.bucket, # 指定目標儲存空間
key=args.key, # 指定對象的鍵名
position=0, # 追加的起始位置，初始為0
body=data1, # 要追加的資料
progress_fn=_progress_fn, # 設定進度回呼函數
))
# 列印第一次追加的結果
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' version id: {result.version_id},'
f' hash crc64: {result.hash_crc64},'
f' next position: {result.next_position},'
)
# 第二次追加資料
result = client.append_object(oss.AppendObjectRequest(
bucket=args.bucket, # 指定目標儲存空間
key=args.key, # 指定對象的鍵名
position=result.next_position, # 從上一次追加的下一個位置開始
body=data2, # 要追加的資料
progress_fn=_progress_fn, # 設定進度回呼函數
))
# 列印第二次追加的結果
print(f'status code: {result.status_code},'
f' request id: {result.request_id},'
f' version id: {result.version_id},'
f' hash crc64: {result.hash_crc64},'
f' next position: {result.next_position},'
)
# 當此指令碼被直接運行時，調用main函數
if __name__ == "__main__":
main()相關文檔
關於追加上傳的完整範例程式碼，請參見append_file.py和append_object.py。