文档首页/ 对象存储服务 OBS/ SDK参考/ Python/ 其他接口(Python SDK)/ 自定义头域(Python SDK)
更新时间：2026-06-30 GMT+08:00
查看PDF
分享

自定义头域(Python SDK)

功能说明

在使用Python SDK调用接口时，您可以为请求添加自定义头域，以发送用户指定的头域。如果您指定的头域属于需要参与签名计算，SDK会自动进行计算。

使用方法为：

调用sdk接口时，在extensionHeaders参数中以字典形式传入指定的头域。

extensionHeaders一般是接口的最后一个参数，为防止参数错位，请以指定形参的方法传参，即extensionHeaders=extensionHeaders

接口约束

OBS支持的Region与Endpoint的对应关系，详细信息请参见地区与终端节点。

代码样例一：单链接限速

本示例用于通过自定义头域实现单链接限速100KB/s下载examplebucket中的objectname对象。

 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
 
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' 
    object_key = 'objectname' 

    # x-obs-traffic-limit为自定义限速功能，单位为bit/s，取值范围[819200, 838860800]即[100KB/s, 100MB/s]，此处以819200（100KB/s）为例 
    extension_headers = {'x-obs-traffic-limit': 819200} 
    # 下载到本地的路径，需包含文件名的全路径 
    download_path = 'localfile' 
    # 单链接限速下载对象 
    resp = obs_client.getObject(bucket_name, object_key, download_path, extensionHeaders=extension_headers) 

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

代码样例二：上传回调

本示例用于通过自定义头域实现普通上传回调。

 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
 
import os
import json
import base64
import traceback
from urllib.parse import quote
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:
    # 指定协议
    protocol = 'http://'

    # 回调地址，url携带特殊字符或中文等需要进行url编码quote(str)
    callback_url1 = protocol + quote('www.example.com/callback1')
    callback_url2 = protocol + quote('www.example.com/中文?key=中文名')
    # 【可选】发起回调请求的Host头域的值，如果不设置，会使用callbackUrl解析出来的Host
    callback_host = 'www.example.com'
    # 发起回调请求的body体
    callback_body = 'key = $(key)&override = $(override)&size = $(size)&bucket = $(bucket)&etag = $(etag)'
    # 组合上传回调配置
    call_back_policy = {'callbackBody': callback_body, 'callbackUrl': callback_url1 + ';' + callback_url2,
                      'callbackHost': callback_host}

    # extensionHeaders自定义头域功能，传参格式为字典
    # 先将上传回调配置转为json字符串，再转为二进制（json.dumps().encode()），再进行base64编码（base64.b64encode()）, base64编码后为bytes类型，需转为str：bytes_data.decode('utf-8')
    extension_headers = {'x-obs-callback': base64.b64encode(json.dumps(call_back_policy).encode()).decode('utf-8')}

    bucket_name = 'your-bucket_name'
    object_key = 'example.txt'
    content = 'Hello OBS'
    # 上传文本并使用上传回调
    resp = obs_client.putContent(bucket_name, object_key, content, extensionHeaders=extension_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())

本示例用于通过自定义头域实现断点续传上传回调。

 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
 
import os
import json
import base64
import traceback
from urllib.parse import quote
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:
    # 指定协议
    protocol = 'http://'

    # 回调地址，url携带特殊字符或中文等需要进行url编码quote(str)
    callback_url1 = protocol + quote('www.example.com/callback1')
    callback_url2 = protocol + quote('www.example.com/中文?key=中文名')
    # 【可选】发起回调请求的Host头域的值，如果不设置，会使用callbackUrl解析出来的Host
    callback_host = 'www.example.com'
    # 发起回调请求的body体
    callback_body = 'key = $(key)&override = $(override)&size = $(size)&bucket = $(bucket)&etag = $(etag)'
    # 组合上传回调配置
    call_back_policy = {'callbackBody': callback_body, 'callbackUrl': callback_url1 + ';' + callback_url2,
                      'callbackHost': callback_host}

    # extensionHeaders自定义头域功能，传参格式为字典
    # 先将上传回调配置转为json字符串，再转为二进制（json.dumps().encode()），再进行base64编码（base64.b64encode()）, base64编码后为bytes类型，需转为str：bytes_data.decode('utf-8')
    extension_headers = {'x-obs-callback': base64.b64encode(json.dumps(call_back_policy).encode()).decode('utf-8')}

    bucket_name = 'your-bucket_name'
    object_key = 'example.txt'
    # 断点续传上传的本地文件路径，需替换为实际文件路径
    file_path = './example.txt'
    # 断点续传上传使用上传回调
    resp = obs_client.uploadFile(bucket_name, object_key, file_path, extensionHeaders=extension_headers)

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

代码样例三：修改对象过期时间

本示例用于通过自定义头域实现修改对象过期时间。 
 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
 
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' 
    object_key = 'objectname' 

    # x-obs-expires为对象过期时间，单位是天，此处以3天为例 
    extensionHeaders = {'x-obs-expires': 3} 
    # 设置对象元数据 
    resp = obs_client.setObjectMetadata(bucket_name, object_key, extensionHeaders=extensionHeaders) 

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

代码样例四：创建桶时开启WORM开关

本示例用于通过自定义头域实现创建桶时开启WORM开关。 
 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
 
import os
import traceback
from obs import ObsClient, CreateBucketHeader

# 推荐通过环境变量获取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: 
    # 创建桶的附加头域，桶的访问控制策略是私有桶，存储类型是标准存储 
    header = CreateBucketHeader(aclControl='PRIVATE', storageClass='STANDARD') 
    # location填写Bucket对应的region, 这里以北京四为例，其他地区请按实际情况填写。 
    location = 'cn-north-4' 
    bucket_name = 'examplebucket' 
    extensionHeaders = {'x-obs-bucket-object-lock-enabled' : 'true'} 
    # 创建桶 
    resp = obs_client.createBucket(bucket_name, header, location, extensionHeaders) 

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

相关链接

使用过程中返回的错误码含义、问题原因及处理措施可参考OBS错误码。