泛化调用

更新时间:

阿里云SDK不仅支持使用云产品SDK提供的特化调用方式调用OpenAPI,还支持使用核心SDK提供的泛化调用方式调用OpenAPI,两种方式的不同请参见泛化调用与特化调用。本文将从获取API信息、安装SDK和代码示例来为您介绍如何使用泛化调用。

获取API信息

访问API文档选择云产品,例如选择云服务器ECS。

  1. 单击云服务器名称下面的获取元数据,在元数据中info.style查看云产品支持的OpenAPI风格(RPC或者ROA)。

    image

    image

    说明

    该位置获取的元数据中包含了云产品的所有API信息,如果您想要查看单个API的元数据,请查看步骤2。

  2. 选择将要调用的API,单击右上角获取元数据

    image

    在元数据中,定义了API支持的网络协议、请求方式、参数及参数位置等信息。如下图所示的RunInstances元数据中:

    • 支持的网络协议有HTTP和HTTPS,建议使用HTTPS。

    • 支持的请求方式有GET和POST,两者请求方式调用结果无任何差异,但GET请求只支持 32 KB 以内的请求包,所以推荐使用POST请求。

    • 支持的参数有RegionId、ImageId等,参数位置在query,表示参数是要拼接在请求URL后面,例如https://ecs.cn-beijing.aliyuncs.com/?ImageId=aliyun_2_1903_x64_20G_alibase_20231221.vhd&InstanceChargeType=PostPaid&InstanceType=ecs.e-c1m1.large&InternetChargeType=PayByTraffic&MinAmount=1&Password=test%401234&RegionId=cn-beijing&SecurityGroupId=sg-2zec0dm6qi66XXXXXXXX&SystemDisk.Category=cloud_essd&SystemDisk.Size=40&VSwitchId=vsw-2ze3aagwn397gXXXXXXXX。

    image

    说明

    元数据中其他支持的内容对签名无影响,这里暂不详细说明。更多元数据的信息,请参见元数据使用指南

安装SDK

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

pip install alibabacloud-tea-openapi

代码示例

示例:调用RPC风格的API

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

import os

from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_tea_util import models as util_models


class Sample:
    @staticmethod
    def main():
        # 从环境变量中获取访问密钥
        config = open_api_models.Config(
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # 设置服务接入点
        config.endpoint = f'ecs-cn-hangzhou.aliyuncs.com'
        client = OpenApiClient(config)
        params = open_api_models.Params(
            style='RPC',  # API风格
            version='2014-05-26',  # API版本号
            action='DescribeRegions',  # API 名称
            method='POST',  # 请求方法
            pathname='/',  # 接口 PATH
            protocol='HTTPS',  # 接口协议,
            auth_type='AK',
            req_body_type='json',  # 接口请求体内容格式,
            body_type='json'  # 接口响应体内容格式,
        )

        # 添加查询参数和请求体
        query = {'InstanceChargeType': 'PrePaid'}
        # 创建API请求对象
        request = open_api_models.OpenApiRequest(
            query=query,
        )
        # 创建运行时选项对象
        runtime = util_models.RuntimeOptions()
        # 发起API调用并打印响应
        response = client.call_api(params, request, runtime)
        print(response)

    @staticmethod
    async def main_async():
        """
        异步主函数,用于演示如何使用异步方式调用阿里云 API。
        """
        # 从环境变量中获取访问密钥
        config = open_api_models.Config(
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        # 设置服务接入点
        config.endpoint = f'ecs-cn-hangzhou.aliyuncs.com'
        client = OpenApiClient(config)
        params = open_api_models.Params(
            style='RPC',  # API风格
            version='2014-05-26',  # API版本号
            action='DescribeRegions',  # API 名称
            method='POST',  # 请求方法
            pathname='/',  # 接口 PATH
            protocol='HTTPS',  # 接口协议,
            auth_type='AK',
            req_body_type='json',  # 接口请求体内容格式,
            body_type='json'  # 接口响应体内容格式,
        )

        # 添加query参数
        query = {'InstanceChargeType': 'PrePaid'}
        # 创建API请求对象
        request = open_api_models.OpenApiRequest(
            query=query,
        )
        # 创建运行时选项对象
        runtime = util_models.RuntimeOptions()
        # 发起异步API调用
        await client.call_api_async(params, request, runtime)


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

示例:调用RESTful(ROA)风格的API

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

import os

from alibabacloud_openapi_util.client import Client as OpenApiUtilClient
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_tea_openapi.client import Client as OpenApiClient
from alibabacloud_tea_util import models as util_models


class Sample:
    @staticmethod
    def main() -> None:
        # 从环境变量中获取访问密钥ID和密钥Secret
        config = open_api_models.Config(
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'cs.cn-qingdao.aliyuncs.com'
        client = OpenApiClient(config)
        params = open_api_models.Params(
            # 接口名称,
            action='DescribeClustersV1',
            # 接口版本,
            version='2015-12-15',
            # 接口协议,
            protocol='HTTPS',
            # 接口 HTTP 方法,
            method='GET',
            auth_type='AK',
            style='ROA',
            # 接口 PATH,
            pathname=f'/api/v1/clusters',
            # 接口请求体内容格式,
            req_body_type='json',
            # 接口响应体内容格式,
            body_type='json'
        )
        # query params
        queries = {'name': 'cluster-demo'}
        request = open_api_models.OpenApiRequest(
            query=OpenApiUtilClient.query(queries)
        )
        # runtime options
        runtime = util_models.RuntimeOptions()
        # 返回值为 Map 类型,可从 Map 中获得三类数据:响应体 body、响应头 headers、HTTP 返回的状态码 statusCode。
        response = client.call_api(params, request, runtime)
        print(response)

    @staticmethod
    async def main_async() -> None:
        # 从环境变量中获取访问密钥ID和密钥Secret
        config = open_api_models.Config(
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
        )
        config.endpoint = f'cs.cn-qingdao.aliyuncs.com'
        client = OpenApiClient(config)
        params = open_api_models.Params(
            # 接口名称,
            action='DescribeClustersV1',
            # 接口版本,
            version='2015-12-15',
            # 接口协议,
            protocol='HTTPS',
            # 接口 HTTP 方法,
            method='GET',
            auth_type='AK',
            style='ROA',
            # 接口 PATH,
            pathname=f'/api/v1/clusters',
            # 接口请求体内容格式,
            req_body_type='json',
            # 接口响应体内容格式,
            body_type='json'
        )
        # query params
        queries = {'name': 'cluster-demo'}
        request = open_api_models.OpenApiRequest(
            query=OpenApiUtilClient.query(queries)
        )
        # runtime options
        runtime = util_models.RuntimeOptions()
        # 返回值为 Map 类型,可从 Map 中获得三类数据:响应体 body、响应头 headers、HTTP 返回的状态码 statusCode。
        await client.call_api_async(params, request, runtime)


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