AI 助理
备案控制台
文档
输入文档关键字查找
首页 对象存储 开发参考 SDK参考 Python 对象/文件 管理文件 拷贝文件

Python拷贝文件

更新时间：
产品详情
我的收藏

本文介绍如何将源Bucket中的文件（Object）复制到同一地域下相同或不同目标Bucket中。

注意事项

  • 本文以华东1（杭州）外网Endpoint为例。如果您希望通过与OSS同地域的其他阿里云产品访问OSS，请使用内网Endpoint。关于OSS支持的Region与Endpoint的对应关系，请参见OSS地域和访问域名。

  • 本文以从环境变量读取访问凭证为例。如何配置访问凭证，请参见配置访问凭证。

  • 本文以OSS域名新建OSSClient为例。如果您希望通过自定义域名、STS等方式新建OSSClient，请参见初始化。

  • 拷贝文件时，您必须拥有源文件的读权限及目标Bucket的读写权限。

  • 拷贝文件时，您需要确保源Bucket和目标Bucket均未设置合规保留策略，否则报错The object you specified is immutable.。

  • 不支持跨地域拷贝。例如不能将华东1（杭州）地域存储空间中的文件拷贝到华北1（青岛）地域。

拷贝小文件

对于小于1 GB的文件，您可以使用简单拷贝。以下代码用于通过简单拷贝将源存储空间srcexamplebucket中的srcexampleobject.txt文件拷贝到目标存储空间destexamplebucket中的destexampleobject.txt文件。

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

# 从环境变量中获取访问凭证。运行本代码示例之前，请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# 填写源Bucket名称，例如srcexamplebucket。
src_bucket_name = 'srcexamplebucket'
# 填写与源Bucket处于同一地域的目标Bucket名称，例如destexamplebucket。
# 当在同一个Bucket内拷贝文件时，请确保源Bucket名称和目标Bucket名称相同。
dest_bucket_name = 'destexamplebucket'

# 填写Bucket所在地域对应的Endpoint。以华东1（杭州）为例，Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# 填写Endpoint对应的Region信息，例如cn-hangzhou。注意，v4签名下，必须填写该参数
region = "cn-hangzhou"
bucket = oss2.Bucket(auth, endpoint, dest_bucket_name, region=region)


# 填写不包含Bucket名称在内源Object的完整路径，例如srcexampleobject.txt。
src_object_name = 'srcexampleobject.txt'
# 填写不包含Bucket名称在内目标Object的完整路径，例如destexampleobject.txt。
dest_object_name = 'destexampleobject.txt'

# headers = dict()
# 指定CopyObject操作时是否覆盖同名目标Object。此处设置为true，表示禁止覆盖同名Object。
# headers['x-oss-forbid-overwrite'] = 'true'
# 指定拷贝的源地址。
# headers[OSS_COPY_OBJECT_SOURCE] = '/example-bucket-by-util/recode-test.txt'
# 如果源Object的ETag值和您提供的ETag相等，则执行拷贝操作，并返回200 OK。
# headers['x-oss-copy-source-if-match'] = '5B3C1A2E053D763E1B002CC607C5****'
# 如果源Object的ETag值和您提供的ETag不相等，则执行拷贝操作，并返回200 OK。
# headers['x-oss-copy-source-if-none-match'] = '5B3C1A2E053D763E1B002CC607C5****'
# 如果指定的时间等于或者晚于文件实际修改时间，则正常拷贝文件，并返回200 OK。
# headers['x-oss-copy-source-if-unmodified-since'] = '2021-12-09T07:01:56.000Z'
# 如果源Object在指定时间后被修改过，则执行拷贝操作。
# headers['x-oss-copy-source-if-modified-since'] = '2021-12-09T07:01:56.000Z'
# 指定设置目标Object元数据的方式。此处设置为COPY，表示复制源Object的元数据到目标Object。
# headers[OSS_METADATA_DIRECTIVE] = 'COPY'
# 指定OSS创建目标Object时使用的服务器端加密算法。
# headers[OSS_SERVER_SIDE_ENCRYPTION] = 'KMS'
# 表示KMS托管的用户主密钥，该参数仅在x-oss-server-side-encryption为KMS时有效。
# headers['x-oss-server-side-encryption-key-id'] = '9468da86-3509-4f8d-a61e-6eab1eac****'
# 指定OSS创建目标Object时的访问权限。此处设置为OBJECT_ACL_PRIVATE，表示只有Object的拥有者和授权用户有该Object的读写权限，其他用户没有权限操作该Object。
# headers[OSS_OBJECT_ACL] = oss2.OBJECT_ACL_PRIVATE
# 指定Object的存储类型。此处设置为BUCKET_STORAGE_CLASS_STANDARD，表示标准存储类型。
# headers['x-oss-storage-class'] = oss2.BUCKET_STORAGE_CLASS_STANDARD
# 指定Object的对象标签，可同时设置多个标签。
# headers[OSS_OBJECT_TAGGING] = 'k1=v1&k2=v2&k3=v3'
# 指定设置目标Object对象标签的方式。此处设置为COPY，表示复制源Object的对象标签到目标Object。
# headers[OSS_OBJECT_TAGGING_COPY_DIRECTIVE] = 'COPY'
# result = bucket.copy_object(src_bucket_name, src_object_name, dest_object_name, headers=headers)

# 将源Bucket中的某个Object拷贝到目标Bucket。
result = bucket.copy_object(src_bucket_name, src_object_name, dest_object_name)

# 查看返回结果的状态。如果返回值为200，表示执行成功。
print('result.status:', result.status)

拷贝大文件

对于大于1 GB的文件，需要使用分片拷贝（UploadPartCopy）。分片拷贝分为三步：

  1. 通过bucket.init_multipart_upload初始化分片拷贝任务。

  2. 通过bucket.upload_part_copy进行分片拷贝。除最后一个分片外，其它分片都要大于100 KB。

  3. 通过bucket.complete_multipart_upload提交分片拷贝任务。

以下代码用于通过分片拷贝将源存储空间srcexamplebucket中的srcexampleobject.txt文件拷贝到目标存储空间destexamplebucket中的destexampleobject.txt文件。

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

# 从环境变量中获取访问凭证。运行本代码示例之前，请确保已设置环境变量OSS_ACCESS_KEY_ID和OSS_ACCESS_KEY_SECRET。
auth = oss2.ProviderAuthV4(EnvironmentVariableCredentialsProvider())

# 填写源Bucket名称，例如srcexamplebucket。
src_bucket_name = 'srcexamplebucket'
# 填写与源Bucket处于同一地域的目标Bucket名称，例如destexamplebucket。
# 当在同一Bucket内拷贝文件时，请确保源Bucket名称和目标Bucket名称相同。
dest_bucket_name = 'destexamplebucket'

# 填写Bucket所在地域对应的Endpoint。以华东1（杭州）为例，Endpoint填写为https://oss-cn-hangzhou.aliyuncs.com。
endpoint = "https://oss-cn-hangzhou.aliyuncs.com"
# 填写Endpoint对应的Region信息，例如cn-hangzhou。注意，v4签名下，必须填写该参数
region = "cn-hangzhou"

# 当在同一Bucket内拷贝文件时，请注释掉该行代码，并将后面的src_bucket改为bucket即可。
bucket = oss2.Bucket(auth, endpoint, dest_bucket_name, region=region)

src_bucket = oss2.Bucket(auth, endpoint, src_bucket_name, region=region)

# 填写不包含Bucket名称在内源Object的完整路径，例如srcexampleobject.txt。
src_object_name = 'srcexampleobject.txt'
# 填写不包含Bucket名称在内目标Object的完整路径，例如destexampleobject.txt。
dest_object_name = 'destexampleobject.txt'
# 获取源文件的大小。当在同一个Bucket内拷贝文件时，请将src_bucket改为bucket。
head_info = src_bucket.head_object(src_object_name)
total_size = head_info.content_length
print('src object size:', total_size)

# determine_part_size方法用来确定分片大小。
part_size = determine_part_size(total_size, preferred_size=100 * 1024)
print('part_size:', part_size)

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

# 逐个上传分片。
part_number = 1
offset = 0
while offset < total_size:
    num_to_upload = min(part_size, total_size - offset)
    end = offset + num_to_upload - 1
    # headers = dict()
    # 指定拷贝的源地址。 
    # headers[OSS_COPY_OBJECT_SOURCE] = '/example-bucket-by-util/recode-test.txt'
    # 指定源Object的拷贝范围。例如设置bytes=0~1023，表示拷贝1~1024字节的内容。
    # headers[OSS_COPY_OBJECT_SOURCE_RANGE] = 'bytes=0~1023'
    # 如果源Object的ETag值和您提供的ETag相等，则执行拷贝操作，并返回200 OK。
    # headers['x-oss-copy-source-if-match'] = '5B3C1A2E053D763E1B002CC6****'
    # 如果源Object的ETag值和您提供的ETag不相等，则执行拷贝操作，并返回200 OK。
    # headers['x-oss-copy-source-if-none-match'] = '5B3C1A2E053D763E1B002CC6****'
    # 如果指定的时间等于或者晚于文件实际修改时间，则正常拷贝文件，并返回200 OK。
    # headers['x-oss-copy-source-if-unmodified-since'] = '2021-12-09T07:01:56.000Z'
    # 如果指定的时间早于文件实际修改时间，则正常拷贝文件，并返回200 OK。
    # headers['x-oss-copy-source-if-modified-since'] = '2021-12-09T07:01:56.000Z'
    # result = bucket.upload_part_copy(src_bucket_name, src_object_name, (offset, end), dest_object_name, upload_id, part_number, headers=headers)
    
    result = bucket.upload_part_copy(src_bucket_name, src_object_name, (offset, end), dest_object_name, upload_id, part_number)
    # 保存part信息。
    parts.append(PartInfo(part_number, result.etag))

    offset += num_to_upload
    part_number += 1

# 完成分片拷贝。
result = bucket.complete_multipart_upload(dest_object_name, upload_id, parts)
# 查看拷贝返回状态。
print('result :', result.status)
# 获取文件元数据。
head_info = bucket.head_object(dest_object_name)
# 查看目标Object大小。
dest_object_size = head_info.content_length
print('dest object size:', dest_object_size)
# 对比源Object和目标Object的大小。
assert dest_object_size == total_size

相关文档

  • 拷贝小文件

    • 关于拷贝小文件的完整示例代码，请参见GitHub示例。

    • 关于拷贝小文件的API接口说明，请参见CopyObject。

  • 拷贝大文件

    关于拷贝大文件的API接口说明，请参见UploadPartCopy。

上一篇：删除文件下一篇：禁止覆盖同名文件
文档推荐