更新时间:2024-03-26 GMT+08:00
分享

自定义头域(Python SDK)

功能说明

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

使用方法为:

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

接口约束

代码样例

本示例用于通过自定义头域实现单链接限速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
from obs import ObsClient
import os
import traceback

# 推荐通过环境变量获取AKSK,这里也可以使用其他外部引入方式传入,如果使用硬编码可能会存在泄露风险。
# 您可以登录访问管理控制台获取访问密钥AK/SK,获取方式请参见https://support.huaweicloud.com/usermanual-ca/ca_01_0003.html。
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 【可选】如果使用临时AKSK和SecurityToken访问OBS,则同样推荐通过环境变量获取
security_token = os.getenv("SecurityToken")
#  server填写Bucket对应的Endpoint, 这里以华北-北京四为例,其他地区请按实际情况填写。
server = "https://obs.cn-north-4.myhuaweicloud.com"
# 创建obsClient实例
# 如果使用临时AKSK和SecurityToken访问OBS,需要在创建实例时通过security_token参数指定securityToken值
obsClient = ObsClient(access_key_id=ak, secret_access_key=sk, server=server)
try:
    bucketName = "examplebucket"
    objectKey = "objectname"
    # x-obs-traffic-limit为自定义限速功能,后面为限速值,单位Bit,取值范围[819200,838860800]即[100KB,100MB],此处以819200(100KB为例)
    extensionHeaders = {'x-obs-traffic-limit': 819200}
    # 下载到本地的路径,localfile为包含本地文件名称的全路径
    downloadPath = 'localfile'
    # 单链接限速下载对象
    resp = obsClient.getObject(bucketName, objectKey, downloadPath, extensionHeaders=extensionHeaders)

    # 返回码为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:
    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
from obs import ObsClient
import os
import traceback
from urllib.parse import quote
import base64
import json

# 推荐通过环境变量获取AKSK,这里也可以使用其他外部引入方式传入,如果使用硬编码可能会存在泄露风险。
# 您可以登录访问管理控制台获取访问密钥AK/SK,获取方式请参见https://support.huaweicloud.com/usermanual-ca/ca_01_0003.html。
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 【可选】如果使用临时AKSK和SecurityToken访问OBS,则同样推荐通过环境变量获取
security_token = os.getenv("SecurityToken")
#  server填写Bucket对应的Endpoint, 这里以华北-北京四为例,其他地区请按实际情况填写。
server = "https://obs.cn-north-4.myhuaweicloud.com"
# 创建obsClient实例
# 如果使用临时AKSK和SecurityToken访问OBS,需要在创建实例时通过security_token参数指定securityToken值
obsClient = ObsClient(access_key_id=ak, secret_access_key=sk, server=server)
try:
    # 指定协议
    protocol = 'http://'

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

    # extensionHeaders自定义头域功能,传参格式为字典
    # 先将上传回调配置转为json字符串,再转为二进制(json.dumps().encode()),再进行base64编码(base64.b64encode()),编码后是二进制,需要转字符串str(b'str', "utf-8")
    extensionHeaders = {'x-obs-callback': str(base64.b64encode(json.dumps(callBackPolicy).encode()), "utf-8")}

    bucketName = 'your-bucketName'
    objectKey = 'example.txt'
    content = 'Hello OBS'
    # 上传文本并使用上传回调
    resp = obsClient.putContent(bucketName, objectKey, content, extensionHeaders=extensionHeaders)

    # 返回码为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:
    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
from obs import ObsClient
import os
import traceback

# 推荐通过环境变量获取AKSK,这里也可以使用其他外部引入方式传入,如果使用硬编码可能会存在泄露风险。
# 您可以登录访问管理控制台获取访问密钥AK/SK,获取方式请参见https://support.huaweicloud.com/usermanual-ca/ca_01_0003.html。
ak = os.getenv("AccessKeyID")
sk = os.getenv("SecretAccessKey")
# 【可选】如果使用临时AKSK和SecurityToken访问OBS,则同样推荐通过环境变量获取
security_token = os.getenv("SecurityToken")
#  server填写Bucket对应的Endpoint, 这里以华北-北京四为例,其他地区请按实际情况填写。
server = "https://obs.cn-north-4.myhuaweicloud.com"
# 创建obsClient实例
# 如果使用临时AKSK和SecurityToken访问OBS,需要在创建实例时通过security_token参数指定securityToken值
obsClient = ObsClient(access_key_id=ak, secret_access_key=sk, server=server)
try:
    bucketName = "examplebucket"
    objectKey = "objectname"
    # x-obs-expires为对象过期时间,单位是天,此处以3天为例
    extensionHeaders = {'x-obs-expires': 3}
    # 设置对象元数据
    resp = obsClient.setObjectMetadata(bucketName, objectKey, 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:
    print('Set Object Metadata Failed')
    print(traceback.format_exc())

相关链接

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

    相关文档

    相关产品