泛化调用

特点

1. 轻量级：仅需安装aliyun-python-sdk-core即可调用所有OpenAPI，无需依赖各产品的独立SDK。

2. 快速迭代兼容性：当云产品尚未提供相应的SDK，或当新API发布但SDK未能及时更新时，可以使用泛化调用。这样，无需等待SDK的更新，即可及时调用最新发布的API接口。

更多介绍，请参见泛化调用与特化调用。

使用说明

在使用泛化调用之前，必须手动获取并配置OpenAPI的版本号、请求路径、参数类型等元数据信息。可以通过查看OpenAPI元数据来获取有关OpenAPI的API风格、请求参数、资源路径等相关信息。

安装核心SDK

在Terminal中执行以下命令安装核心SDK。

pip install aliyun-python-sdk-core

调用OpenAPI

初始化请求客户端

通过在aliyunsdkcore包中创建client模块以初始化请求客户端，并通过该Client发起OpenAPI调用。此处仅列举使用AccessKey（简称：AK）初始化客户端的方式，更多初始化方式请参见管理访问凭据。

import os
from aliyunsdkcore.client import AcsClient

# 使用AccessKey直接初始化。
client = AcsClient(
    os.environ['ALIYUN_ACCESS_KEY_ID'],
    os.environ['ALIYUN_ACCESS_KEY_SECRET'],
    region_id='cn-hangzhou',
    # verify=False,  # 忽略对 SSL 证书的验证
    # proxy={'http': 'http://127.0.0.1:9898'}, # 设置代理
    # proxy={'https': 'http://<user>:<password>@127.0.0.1:8989'},
    # connect_timeout=10,  # 连接超时
    # timeout=15  # 读超时时间
)

配置OpenAPI信息及请求参数

通过CommonRequest来配置OpenAPI所需的公共请求参数及接口请求参数。关于更多公共请求参数的介绍及使用请参见进阶配置。

# 2.创建CommonRequest对象并配置OpenAPI基本信息和请求参数。
request = CommonRequest()

# 2.1 配置公共请求参数
request.set_domain('ecs-cn-hangzhou.aliyuncs.com')  # API服务地址
request.set_version('2014-05-26')  # API版本号
request.set_action_name('DescribeInstanceStatus')  # API名称，当调用RPC风格API时，必须配置set_action_name()指定接口名称
request.set_method('POST')  # API请求方式
request.set_protocol_type('HTTPS')  # 请求协议：HTTPS或HTTP，建议使用HTTPS。
# request.set_uri_pattern('/')  # API资源路径，ROA接口必传。RPC接口禁止设置该参数。

# 2.2 配置接口请求参数
# 场景一：Query参数通过add_query_param(self, k, v)设置
InstanceIds = [
    "i-bp1axhql4dqXXXXXXXX",
    "i-bp124uve8zqXXXXXXXX"
]
for depth1 in range(len(InstanceIds)):
    request.add_query_param('InstanceId.' + str(depth1 + 1), InstanceIds[depth1])
request.add_query_param('PageNumber', '1')
request.add_query_param('PageSize', '30')

# 场景二：body参数通过add_body_params(self, k, v)设置
# request.add_body_params('key1', 'value1')
# request.add_body_params('key2', 'value2')
# request.add_body_params('key3', 'value3')

# 场景三：文件上传set_content(self, content)，content为字节数组类型
# file_path = "<FILE_PATH>" # <FILE_PATH>需替换为实际文件路径
# with open(file_path, 'rb') as file:
#    读取图片内容为字节数组
#    ByteArray = file.read()
# request.set_content(ByteArray)

发起请求

通过client调用do_action_with_exception发起请求，参数为前一步构造的CommonRequest实例。

# 发起请求
response = client.do_action_with_exception(request)
# 返回值是 bytes 类型，包含RequestId以及OpenAPI的返回参数
print(f"response：\n{response}")
# 解析响应内容（JSON 格式）
response_json = json.loads(response.decode('utf-8'))  # 转换为Python字典
print("RequestId：", response_json["RequestId"])  # 请求ID

代码示例

示例：调用RPC风格的API

以调用ECS的DescribeInstanceStatus接口为例，展示如何使用泛化调用方式。

import os
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest


class Sample:

    @staticmethod
    def main():
        # 使用AccessKey直接初始化。
        client = AcsClient(os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'], os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'],
                           'cn-hangzhou')
        # 创建请求对象
        request = CommonRequest()
        request.set_domain('ecs-cn-hangzhou.aliyuncs.com')  # API服务地址
        request.set_version('2014-05-26')  # API版本号
        request.set_action_name('DescribeRegions')  # API名称，当调用RPC风格API时，必须配置set_action_name()指定接口名称
        request.set_method('POST')  # API请求方式
        request.set_protocol_type('HTTPS')  # 请求协议：HTTPS或HTTP，建议使用HTTPS。
        InstanceIds = [
            "i-bp1axhql4dqXXXXXXXX",
            "i-bp124uve8zqXXXXXXXX"
        ]
        for depth1 in range(len(InstanceIds)):
            request.add_query_param('InstanceId.' + str(depth1 + 1), InstanceIds[depth1])
        request.add_query_param('PageNumber', '1')
        request.add_query_param('PageSize', '30')
        # 发起请求并处理响应
        response = client.do_action_with_exception(request)
        print(response)


if __name__ == '__main__':
    Sample.main()
     

示例：调用RESTful（ROA）风格的API

以调用容器服务查询集群列表信息DescribeClustersV1接口为例，展示如何使用泛化调用。

import os
from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.request import CommonRequest


class Sample:

    @staticmethod
    def main():
        # 使用AccessKey直接初始化。
        client = AcsClient(
            os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET'],
            'cn-hangzhou')
        # 创建请求对象
        request = CommonRequest()
        request.set_domain('cs.aliyuncs.com')  # API服务地址
        request.set_version('2015-12-15')  # API版本号
        request.set_uri_pattern(
            f'/api/v1/clusters')  # API资源路径，当调用ROA风格API时，必须配置set_uri_pattern()指定完整资源路径。从OpenAPI元数据中data.path获取资源路径。
        request.set_method('GET')  # API请求方法
        request.add_query_param('name', 'cluster-demo')  # 配置请求参数
        # 发起请求并处理响应
        response = client.do_action_with_exception(request)
        print(response)


if __name__ == '__main__':
    Sample.main()