All Products
Search

This Product

    Object Storage Service:Prevent overwrites of same-name objects using OSS SDK for Python 1.0

    Document Center

    Object Storage Service:Prevent overwrites of same-name objects using OSS SDK for Python 1.0

    Last Updated:Mar 20, 2026

    By default, OSS overwrites an existing object when you upload an object with the same key. Set the x-oss-forbid-overwrite request header to true to reject the write and return an error instead. This header applies to simple upload, object copy, and multipart upload.

    Usage notes

    • The examples in this topic use the public endpoint for the China (Hangzhou) region. To access OSS from another Alibaba Cloud service in the same region, use an internal endpoint instead. For more information, see Regions and endpoints.

    • Access credentials in these examples are read from environment variables. For configuration details, see Configure access credentials.

    • These examples create an OSSClient instance using an OSS endpoint. To use a custom domain name or Security Token Service (STS), see Initialization.

    Simple upload

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

# Read access credentials from environment variables.
# Set OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET before running this example.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint for the region where your bucket is located.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region. Required for V4 signatures.
region = "cn-hangzhou"

bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

# Set x-oss-forbid-overwrite to true to reject the upload if an object
# with the same key already exists.
headers = {'x-oss-forbid-overwrite': 'true'}
result = bucket.put_object('yourObjectName', 'content of object', headers=headers)

print('http status: {0}'.format(result.status))
print('request_id: {0}'.format(result.request_id))
print('ETag: {0}'.format(result.etag))
print('date: {0}'.format(result.headers['date']))

    Copy an object

    Copy a small object

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

# Read access credentials from environment variables.
# Set OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET before running this example.
auth = oss2.ProviderAuth(EnvironmentVariableCredentialsProvider())

bucket = oss2.Bucket(auth, 'https://oss-cn-hangzhou.aliyuncs.com', 'yourBucketName')

# Set x-oss-forbid-overwrite to true to reject the copy if an object
# with the same key already exists at the destination.
headers = {'x-oss-forbid-overwrite': 'true'}
bucket.copy_object('yourSourceBucketName', 'yourSourceObjectName', 'yourDestinationObjectName', headers=headers)

    Copy a large object (multipart copy)

    For large objects, use multipart copy. Set the header on both init_multipart_upload and complete_multipart_upload.

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

# Read access credentials from environment variables.
# Set OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET before running this example.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint for the region where your bucket is located.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region. Required for V4 signatures.
region = "cn-hangzhou"

bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

src_object = 'yourSourceObjectName'
dst_object = 'yourDestinationObjectName'

total_size = bucket.head_object(src_object).content_length
part_size = determine_part_size(total_size, preferred_size=100 * 1024)

# Set x-oss-forbid-overwrite on InitiateMultipartUpload.
headers = {'x-oss-forbid-overwrite': 'true'}
upload_id = bucket.init_multipart_upload(dst_object, headers=headers).upload_id
parts = []

# Copy parts sequentially.
part_number = 1
offset = 0
while offset < total_size:
    num_to_upload = min(part_size, total_size - offset)
    byte_range = (offset, offset + num_to_upload - 1)

    result = bucket.upload_part_copy(bucket.bucket_name, src_object, byte_range, dst_object, upload_id, part_number)
    parts.append(PartInfo(part_number, result.etag))

    offset += num_to_upload
    part_number += 1

# Set x-oss-forbid-overwrite on CompleteMultipartUpload as well.
headers = {'x-oss-forbid-overwrite': 'true'}
bucket.complete_multipart_upload(dst_object, upload_id, parts, headers=headers)

    Multipart upload

    Set the header on both init_multipart_upload and complete_multipart_upload.

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

# Read access credentials from environment variables.
# Set OSS_ACCESS_KEY_ID and OSS_ACCESS_KEY_SECRET before running this example.
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# Specify the endpoint for the region where your bucket is located.
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# Specify the region. Required for V4 signatures.
region = "cn-hangzhou"

bucket = oss2.Bucket(auth, endpoint, "yourBucketName", region=region)

key = 'yourObjectName'
filename = 'yourLocalFile'

total_size = os.path.getsize(filename)
part_size = determine_part_size(total_size, preferred_size=100 * 1024)

# Set x-oss-forbid-overwrite on InitiateMultipartUpload.
headers = {'x-oss-forbid-overwrite': 'true'}
upload_id = bucket.init_multipart_upload(key, headers=headers).upload_id
parts = []

# Upload parts sequentially.
# SizedFileAdapter generates a new file object and recalculates the starting
# read position for each part.
with open(filename, 'rb') as fileobj:
    part_number = 1
    offset = 0
    while offset < total_size:
        num_to_upload = min(part_size, total_size - offset)
        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

# Set x-oss-forbid-overwrite on CompleteMultipartUpload as well.
headers = {'x-oss-forbid-overwrite': 'true'}
bucket.complete_multipart_upload(key, upload_id, parts, headers=headers)

# Verify the upload.
with open(filename, 'rb') as fileobj:
    assert bucket.get_object(key).read() == fileobj.read()

    What's next