Python调用示例
更新时间 2026-07-16 15:51:46
最近更新时间: 2026-07-16 15:51:46
此示例仅适用于 Python 3.6 以上版本
# encoding: utf-8
import requests
import json
import hashlib
import base64
import hmac
import datetime
import uuid
METHOD_GET = 'GET'
METHOD_POST = 'POST'
METHOD_HEAD = 'HEAD'
METHOD_PATCH = 'PATCH'
METHOD_PUT = 'PUT'
METHOD_DELETE = 'DELETE'
# 用户天翼云AK/SK
AK = 'xxx'
SK = 'xxx'
def hmac_sha256(secret, data):
return hmac.new(secret, data, digestmod=hashlib.sha256).digest()
def get_sorted_str(data):
"""
鉴权用的参数整理
:param data: dict 需要整理的参数
:return: str
"""
sorted_data = sorted(data.items(), key=lambda item: item[0])
str_list = map(lambda x_y: '%s=%s' % (x_y[0], x_y[1]), sorted_data)
return '&'.join(str_list)
def build_sign(query_params, body_params, eop_date, request_uuid):
"""
计算鉴权字段
:param query_params: dict get请求中的参数
:param body_params: dict post请求中的参数
:param eop_date: str 请求时间,格式为:'%Y%m%dT%H%M%SZ'
:return: str
"""
# 步骤一:构造代签字符串sigture
body_str = json.dumps(body_params)
body_digest = hashlib.sha256(body_str.encode('utf-8')).hexdigest()
header_str = 'ctyun-eop-request-id:%s\neop-date:%s\n' % (request_uuid, eop_date) # 请求头中必要的两个参数
query_str = get_sorted_str(query_params) # url中的参数,或get参数
signature_str = '%s\n%s\n%s' % (header_str, query_str, body_digest)
# 步骤二:构造动态秘钥kdate
sign_date = eop_date.split('T')[0]
k_time = hmac_sha256(SK.encode('utf-8'), eop_date.encode('utf-8'))
k_ak = hmac_sha256(k_time, AK.encode('utf-8'))
k_date = hmac_sha256(k_ak, sign_date.encode('utf-8'))
# 步骤三:构造signature
signature_base64 = base64.b64encode(hmac_sha256(k_date, signature_str.encode('utf-8'))).decode('ascii')
# 步骤四:构造Eop-Authorization
sign_header = f'{AK} Headers=ctyun-eop-request-id;eop-date Signature={signature_base64}'
return sign_header
def get_sign_headers(query_params, body):
"""
获取鉴权用的请求头参数
:param query_params: dict get请求中的参数
:param body: dict post请求中的参数
:return:
"""
now = datetime.datetime.now()
eop_date = datetime.datetime.strftime(now, '%Y%m%dT%H%M%SZ')
request_uuid = str(uuid.uuid1())
headers = { # 三个鉴权用的参数
'eop-date': eop_date,
'ctyun-eop-request-id': request_uuid,
'Eop-Authorization': build_sign(query_params=query_params, body_params=body, eop_date=eop_date,
request_uuid=request_uuid),
}
return headers
def get(url, params=None, header_params=None, get_body=None):
return execute(url, method=METHOD_GET, params=params, header_params=header_params, get_body=get_body)
def post(url, params=None, header_params=None):
return execute(url, method=METHOD_POST, params=params, header_params=header_params)
def put(url, params=None, header_params=None):
return execute(url, method=METHOD_PUT, params=params, header_params=header_params)
def delete(url, params=None, header_params=None):
return execute(url, method=METHOD_DELETE, params=params, header_params=header_params)
def head(url, params=None, header_params=None):
return execute(url, method=METHOD_HEAD, params=params, header_params=header_params)
def patch(url, params=None, header_params=None):
return execute(url, method=METHOD_PATCH, params=params, header_params=header_params)
def print_log(log_info):
now = datetime.datetime.now()
print('[%s]: %s' % (str(now), log_info))
def execute(url, method, params=None, header_params=None, get_body=None):
get_body = get_body or {}
params = params or {}
header_params = header_params or {}
query_params, body = (params, get_body) if method == METHOD_GET else ({}, params)
headers = get_sign_headers(query_params, body)
headers.update(header_params)
print_log('url: %s' % url)
print_log('请求方式: %s' % method)
print_log('请求头: %s' % headers)
print_log('请求参数: %s' % params)
if method == METHOD_GET:
res = requests.get(url, params=params, headers=headers, verify=False, json=get_body)
elif method == METHOD_POST:
res = requests.post(url, json=params, headers=headers, verify=False)
elif method == METHOD_PUT:
res = requests.put(url, json=params, headers=headers, verify=False)
elif method == METHOD_DELETE:
res = requests.delete(url, json=params, headers=headers, verify=False)
elif method == METHOD_HEAD:
res = requests.head(url, json=params, headers=headers, verify=False)
elif method == METHOD_PATCH:
res = requests.patch(url, json=params, headers=headers, verify=False)
else:
raise ValueError(f"Unsupported method: {method}")
print_log('返回状态码: %s' % res.status_code)
print_log('返回: %s' % res.text)
return res
# 下面以查询所有对象存储资源池为例
def list_regions():
url = "https://zos-global.ctapi.ctyun.cn/v4/oss/list-regions"
params = {}
get(url, params=params, header_params={})
if __name__ == '__main__':
list_regions()