文档首页/ 对象存储服务 OBS/ SDK参考/ Python/ 对象相关接口(Python SDK)/ 一致性校验(Python SDK)/ 使用MD5校验文件一致性(Python SDK)
更新时间：2026-06-30 GMT+08:00
查看PDF
分享

使用MD5校验文件一致性(Python SDK)

您可以通过以下接口在上传对象时使用md5校验文件一致性：

代码样例一：文本上传

本示例用于上传文本对象时使用MD5校验文件一致性。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
 
import base64
import hashlib
import os
import traceback

from obs import ObsClient
from obs import PutObjectHeader

# 推荐通过环境变量获取AK/SK，这里也可以使用其他外部引入方式传入，如果使用硬编码可能会存在泄露风险
# 您可以登录访问管理控制台获取访问密钥AK/SK，获取方式请参见https://support.huaweicloud.com/intl/zh-cn/usermanual-ca/ca_01_0003.html
# 运行本代码示例之前，请确保已设置环境变量AccessKeyID和SecretAccessKey
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 如需使用临时AK/SK和SecurityToken访问OBS，则同样推荐通过环境变量获取
# security_token = os.getenv("SecurityToken")
# server填写Bucket对应的Endpoint，这里以中国-香港为例，其他地区请按实际情况填写
server = "https://obs.ap-southeast-1.myhuaweicloud.com"
# 创建obsClient实例
# 如果使用临时AK/SK和SecurityToken访问OBS，需要在创建实例时通过security_token参数指定securityToken值
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    bucket_name = "examplebucket" 
    object_key = "objectname" 
    # 设置待上传的文本内容 
    content = 'Hello OBS' 
    # 上传对象的附加头域 
    headers = PutObjectHeader() 
    # 获取MD5值 
    md5 = hashlib.md5(content.encode('utf-8')).digest() 
    # 对MD5值进行Base64编码 
    md5_base64 = base64.b64encode(md5).decode('utf-8') 
    headers.md5 = md5_base64 
    # 上传文本对象 
    resp = obs_client.putContent(bucket_name, object_key, content, headers=headers) 
    # 返回码为2xx时，接口调用成功，否则接口调用失败 
    if resp.status < 300: 
        print('Put Content Succeeded') 
        print('requestId:', resp.requestId) 
        print('etag:', resp.body.etag) 
    else: 
        print('Put Content Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
        print('errorMessage:', resp.errorMessage) 
except Exception: 
    print('Put Content Failed') 
    print(traceback.format_exc())

代码样例二：文件上传

本示例用于上传单个文件时使用MD5校验文件一致性。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
 
import base64
import hashlib
import os
import traceback

from obs import ObsClient
from obs import PutObjectHeader

# 推荐通过环境变量获取AK/SK，这里也可以使用其他外部引入方式传入，如果使用硬编码可能会存在泄露风险
# 您可以登录访问管理控制台获取访问密钥AK/SK，获取方式请参见https://support.huaweicloud.com/intl/zh-cn/usermanual-ca/ca_01_0003.html。
# 运行本代码示例之前，请确保已设置环境变量AccessKeyID和SecretAccessKey
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 如需使用临时AK/SK和SecurityToken访问OBS，则同样推荐通过环境变量获取
# security_token = os.getenv("SecurityToken")
# server填写Bucket对应的Endpoint，这里以中国-香港为例，其他地区请按实际情况填写
server = "https://obs.ap-southeast-1.myhuaweicloud.com"
# 创建obsClient实例
# 如果使用临时AK/SK和SecurityToken访问OBS，需要在创建实例时通过security_token参数指定securityToken值
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    bucket_name = "examplebucket" 
    # 对象名，即上传后的文件名 
    object_key = "objectname" 
    # 上传对象的附加头域 
    headers = PutObjectHeader() 
    # 待上传文件的完整路径，如aa/bb.txt 
    file_path = 'localfile' 
    # 【可选】待上传对象的md5值 
    with open(file_path, 'rb') as f: 
        md5_obj = hashlib.md5() 
        while True: 
            data = f.read() 
            if not data: 
                break 
        md5_obj.update(data) 
        # 获取MD5值 
        md5 = md5_obj.digest() 
        # 对MD5值进行Base64加密 
        md5_base64 = base64.b64encode(md5) 
    headers.md5 = md5_base64 
    # 文件上传 
    resp = obs_client.putFile(bucket_name, object_key, file_path, headers=headers) 
    # 返回码为2xx时，接口调用成功，否则接口调用失败 
    if resp.status < 300: 
        print('Put File Succeeded') 
        print('requestId:', resp.requestId) 
        print('etag:', resp.body.etag) 
        print('versionId:', resp.body.versionId) 
        print('storageClass:', resp.body.storageClass) 
    else: 
        print('Put File Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
except Exception: 
    print('Put File Failed') 
    print(traceback.format_exc())

代码样例三：追加上传

本示例用于追加上传对象时使用MD5校验文件一致性。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
 
import base64
import hashlib
import os
import traceback

from obs import AppendObjectContent, AppendObjectHeader
from obs import ObsClient

# 推荐通过环境变量获取AK/SK，这里也可以使用其他外部引入方式传入，如果使用硬编码可能会存在泄露风险
# 您可以登录访问管理控制台获取访问密钥AK/SK，获取方式请参见https://support.huaweicloud.com/intl/zh-cn/usermanual-ca/ca_01_0003.html
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 如需使用临时AK/SK和SecurityToken访问OBS，则同样推荐通过环境变量获取
# security_token = os.getenv("SecurityToken")
# server填写Bucket对应的Endpoint，这里以中国-香港为例，其他地区请按实际情况填写
server = "https://obs.ap-southeast-1.myhuaweicloud.com"
# 创建obsClient实例
# 如果使用临时AK/SK和SecurityToken访问OBS，需要在创建实例时通过security_token参数指定securityToken值
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    headers = AppendObjectHeader() 
    # 追加上传的消息体 
    content = AppendObjectContent() 
    # 设置追加内容 
    content.content = 'Hello OBS' 
    # 设置追加内容起始位置，此处以对象的0个字节为起始位置 
    content.position = 0 
    # 第一次调用追加上传时，如果已存在同名的普通对象，则抛出异常（状态码为409） 
    bucket_name = "examplebucket" 
    # 追加上传的对象名 
    object_key = "objectname" 
    # 获取MD5值 
    md5 = hashlib.md5(content.content.encode('utf-8')).digest() 
    # 对MD5值进行Base64编码 
    md5_base64 = base64.b64encode(md5).decode('utf-8') 
    headers.md5 = md5_base64 
    # 追加上传 
    resp = obs_client.appendObject(bucket_name, object_key, content, headers=headers) 

    # 返回码为2xx时，接口调用成功，否则接口调用失败 
    if resp.status < 300: 
        print('Append Object Succeeded') 
        print('requestId:', resp.requestId) 
        print('crc64:', resp.body.crc64) 
        print('nextPosition:', resp.body.nextPosition) 
    else: 
        print('Append Object Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
        print('errorMessage:', resp.errorMessage) 
except Exception: 
    print('Append Object Failed') 
    print(traceback.format_exc())

代码样例四：上传段

本示例用于展示上传段时使用MD5校验段数据一致性。

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
 
import os
import traceback

from obs import ObsClient

# 推荐通过环境变量获取AK/SK，这里也可以使用其他外部引入方式传入，如果使用硬编码可能会存在泄露风险
# 您可以登录访问管理控制台获取访问密钥AK/SK，获取方式请参见https://support.huaweicloud.com/intl/zh-cn/usermanual-ca/ca_01_0003.html
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 如需使用临时AK/SK和SecurityToken访问OBS，则同样推荐通过环境变量获取
# security_token = os.getenv("SecurityToken")
# server填写Bucket对应的Endpoint，这里以中国-香港为例，其他地区请按实际情况填写
server = "https://obs.ap-southeast-1.myhuaweicloud.com" 
# 创建obsClient实例
# 如果使用临时AK/SK和SecurityToken访问OBS，需要在创建实例时通过security_token参数指定securityToken值
obs_client = ObsClient(access_key_id=ak, secret_access_key=sk, server=server) 
try: 
    bucket_name = "examplebucket" 
    # 上传到obs桶里的对象名称 
    object_key = "objectname" 
    # 段号，取值范围：1~10000 
    part_number = "your part_number" 
    # 分段上传任务的ID 
    upload_id = "your uploadid" 
    # 待上传段的内容str或readable object 
    object = 'Hello OBS' 
    # 指明object字段是否代表文件路径，默认为False 
    is_file = False 
    # 源文件中某一分段的起始偏移大小，默认值为0， 单位为字节 
    offset = 0 
    # 源文件中某一分段的大小，默认值为文件大小减去offset的剩下字节数，单位为字节 
    part_size = 9 * 1024 * 1024 
    # 是否自动计算待上传数据的MD5值，默认为False 
    is_attach_md5 = True 
    # 通过分段上传任务的ID，上传段到指定桶中 
    resp = obs_client.uploadPart(bucket_name, object_key, part_number, upload_id, object, is_file, part_size, 
                                offset, isAttachMd5 = is_attach_md5) 

    # 返回码为2xx时，接口调用成功，否则接口调用失败 
    if resp.status < 300: 
        print('Upload Part Succeeded') 
        print('requestId:', resp.requestId) 
        print('etag:', resp.body.etag) 
    else: 
        print('Upload Part Failed') 
        print('requestId:', resp.requestId) 
        print('errorCode:', resp.errorCode) 
        print('errorMessage:', resp.errorMessage) 
except Exception: 
    print('Upload Part Failed') 
    print(traceback.format_exc())