How to use OSS SDK for Python V2 to copy files - Object Storage Service

The Copier is a high-level copy interface in OSS SDK for Python V2 that automatically selects between CopyObject and multipart copy (UploadPartCopy) based on object size. Use it instead of calling those APIs directly — the Copier handles operation selection, part splitting, and concurrency for you.

Prerequisites

Before you begin, make sure you have:

Read permission on the source object
Read and write permissions on the destination bucket
OSS SDK for Python V2 installed (alibabacloud_oss_v2)

How it works

The Copier picks the copy operation based on object size and the multipart_copy_threshold setting:

Object size	Operation used	Notes
Below threshold (default: 200 MiB)	`CopyObject`	Single-request copy; supports `x-oss-metadata-directive` and `x-oss-tagging-directive`. Suitable only for objects smaller than 5 GiB.
At or above threshold	`UploadPartCopy` (multipart copy)	Parallel copy in parts; `x-oss-metadata-directive` and `x-oss-tagging-directive` are not supported — specify metadata and tags explicitly in the request

Cross-region copy is not supported. Source and destination buckets must be in the same region.

If a retention policy is configured on the source or destination bucket, the copy fails with: The object you specified is immutable.

Copy an object

The following example copies an object from a source bucket to a destination bucket using default settings.

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="copier 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 destination 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 destination object.', required=True)
parser.add_argument('--source_key', help='The name of the source object.', required=True)
parser.add_argument('--source_bucket', help='The name of the source bucket.', required=True)


def main():
    args = parser.parse_args()

    # Load AccessKey ID and AccessKey secret from environment variables.
    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    # Create a Copier instance with default settings.
    copier = client.copier()

    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,                    # Destination bucket
            key=args.key,                          # Destination object key
            source_bucket=args.source_bucket,      # Source bucket
            source_key=args.source_key             # Source object key
        )
    )

    print(vars(result))


if __name__ == "__main__":
    main()

The sample code uses region cn-hangzhou and a public endpoint by default. If you access OSS from another Alibaba Cloud service in the same region, use an internal endpoint. For endpoint mappings, see Regions and endpoints.

API reference

`client.copier(**kwargs) -> Copier`

Creates a Copier instance. Pass configuration options as keyword arguments to apply them globally for all copy calls on that instance.

`copier.copy(request, **kwargs) -> CopyResult`

Copies an object. Pass configuration options as keyword arguments to override instance-level settings for that specific call.

`CopyObjectRequest` parameters

Parameter	Type	Description
`bucket`	str	Name of the destination bucket
`key`	str	Key of the destination object
`source_bucket`	str	Name of the source bucket
`source_key`	str	Key of the source object
`forbid_overwrite`	str	Specifies whether to overwrite a destination object with the same key during the CopyObject operation.
`tagging`	str	Tags to apply to the destination object. Format: `TagA=A&TagB=B`
`tagging_directive`	str	How to handle tags on the destination object. `Copy` (default): copy tags from the source. `Replace`: use the tags specified in `tagging`.
`progress_fn`	callable	Progress callback. Called periodically with `(bytes_transferred_in_this_chunk, total_bytes_written, total_bytes)`.

Configuration options

Set these at the Copier level (client.copier(...)) or per call (copier.copy(request, ...)). Per-call settings override instance-level settings.

Parameter	Type	Default	Description
`part_size`	int	64 MiB	Size of each part in a multipart copy
`parallel_num`	int	3	Number of concurrent part copy tasks. This is a per-call limit, not a global limit.
`multipart_copy_threshold`	int64	200 MiB	Object size above which multipart copy is used
`leave_parts_on_error`	bool	`False`	Specifies whether to retain the copied parts if the copy fails. By default, the copied parts are not retained.
`disable_shallow_copy`	bool	`False`	Specifies whether to disable shallow copy. By default, shallow copy is used.

Scenarios

Set part size and concurrency

Override part_size, parallel_num, and leave_parts_on_error per copy call:

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="copier 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 destination 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 destination object.', required=True)
parser.add_argument('--source_key', help='The name of the source object.', required=True)
parser.add_argument('--source_bucket', help='The name of the source bucket.', required=True)


def main():
    args = parser.parse_args()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    copier = client.copier()

    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,
            key=args.key,
            source_bucket=args.source_bucket,
            source_key=args.source_key
        ),
        part_size=1 * 1024 * 1024,    # 1 MiB per part
        parallel_num=5,                # 5 concurrent part copy tasks
        leave_parts_on_error=True,     # Keep parts on failure
    )

    print(vars(result))


if __name__ == "__main__":
    main()

Track copy progress

Pass a progress_fn callback to CopyObjectRequest to monitor copy progress:

import argparse
import alibabacloud_oss_v2 as oss

parser = argparse.ArgumentParser(description="copier 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 destination 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 destination object.', required=True)
parser.add_argument('--source_key', help='The name of the source object.', required=True)
parser.add_argument('--source_bucket', help='The name of the source bucket.', required=True)


def main():
    args = parser.parse_args()

    credentials_provider = oss.credentials.EnvironmentVariableCredentialsProvider()

    cfg = oss.config.load_default()
    cfg.credentials_provider = credentials_provider
    cfg.region = args.region
    if args.endpoint is not None:
        cfg.endpoint = args.endpoint

    client = oss.Client(cfg)

    def _progress_fn(n, written, total):
        # n: bytes transferred in this chunk
        # written: total bytes written so far
        # total: total object size
        rate = int(100 * (float(written) / float(total)))
        print(f'\rCopy progress: {rate}% ', end='')

    copier = client.copier()

    result = copier.copy(
        oss.CopyObjectRequest(
            bucket=args.bucket,
            key=args.key,
            source_bucket=args.source_bucket,
            source_key=args.source_key,
            progress_fn=_progress_fn,    # Attach the progress callback
        )
    )

    print(vars(result))


if __name__ == "__main__":
    main()

What's next

For a full explanation of the Copier design and advanced options, see the Developer Guide.
For complete runnable sample code, see copier.py.