自定义头域(Python SDK)

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())

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())