Python V2.0 SDK泛化调用_阿里云SDK(Alibaba Cloud SDK)-阿里云帮助中心

阿里云SDK支持一种通用的方式调用OpenAPI，此方式被称为泛化调用。本文将为您详细介绍如何使用泛化调用访问OpenAPI。

特点

轻量：仅需安装阿里云核心SDK，无需额外安装云产品SDK。

简单：只需构造通用的请求参数对象，然后利用通用请求客户端调用通用函数发起请求，调用结果也以通用格式返回。

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

使用说明

使用泛化调用时，建议先查看OpenAPI元数据，获取OpenAPI的API风格、请求参数、资源路径等信息。

安装核心SDK

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

pip install alibabacloud-tea-openapi

调用OpenAPI

1. 初始化请求客户端

通过创建alibabacloud_tea_openapi包中client模块的Client类初始化请求客户端，并通过该Client发起OpenAPI调用。在初始化客户端时，也支持使用Credentials工具，关于Credentials工具的更多信息，请参见管理访问凭据。

重要

客户端实例是线程安全的。在实际开发中建议使用单例模式创建客户端，确保在整个应用程序生命周期中针对相同的访问凭据和Endpoint仅初始化一个客户端对象实例。

import os

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

# 从环境变量中获取访问密钥
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'
# config.protocol = 'HTTPS' # 通过 HTTPS 协议发送请求
# config.http_proxy = 'http://127.0.0.1:9898'
# config.https_proxy='http://user:password@127.0.0.1:8989'
# config.read_timeout = 10000,  # 读超时时间 单位毫秒(ms)
# config.connect_timeout = 5000  # 连接超时 单位毫秒(ms)
client = OpenApiClient(config)


# 使用Credentials工具
# crendconfig = credentialConfig()
# crendconfig.type = 'access_key'
# crendconfig.access_key_id = os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID']
# crendconfig.access_key_secret = os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
# 
# credentialsClient = CredClient(crendconfig)
# config = open_api_models.Config()
# config.credential = credentialsClient
# config.endpoint = f'ecs.cn-hangzhou.aliyuncs.com'
# client = OpenApiClient(config)

2. 配置OpenAPI信息

通过alibabacloud_tea_openapi包中models模块的Params类配置OpenAPI的基本信息，比如OpenAPI的风格、API版本、请求方式等信息。以调用DescribeInstanceTypeFamilies接口为例：

params = open_api_models.Params(
  style='RPC',  # API风格：RPC或ROA。
  version='2014-05-26',  # API版本号
  action='DescribeInstanceTypeFamilies',  # API 名称
  method='POST',  # 请求方法
  pathname='/',  # API资源路径，RPC接口默认"/"，ROA接口从OpenAPI元数据中data.path获取资源路径。
  protocol='HTTPS',  # 请求协议：HTTPS或HTTP，建议使用HTTPS。
  auth_type='AK',  # 认证类型，默认即可。当OpenAPI支持匿名请求时，您可以传入 Anonymous 发起匿名请求。
  req_body_type='json',  # 请求body的类型，支持byte、json、formData。
  body_type='json'  # 返回数据格式，支持json。
)

3. 配置请求参数

通过alibabacloud_tea_openapi包中models模块的OpenApiRequest类配置请求参数，请求参数支持通过query、body或stream传参，如何选择传参方式可根据元数据中的介绍选择，例如DescribeInstanceTypeFamilies的请求参数RegionId在元数据中信息为{"name":"RegionId","in":"query",...}}，其中"in":"query"表示RegionId通过Query传递。

传参方式	描述
query	请求参数显示`"in":"query"`时，通过query传递参数。
body	请求参数显示`"in":"body"`或`"in": "formData"`时，通过Body传递参数。在使用请求体（body）传递参数时，需要根据请求体的类型设置`reqBodyType`的值。
stream	在需要上传文件的场景，可通过stream传递文件流。

from alibabacloud_openapi_util.client import Client as OpenApiUtilClient

# 场景一：设置查询参数（query）
query = {'RegionId': 'cn-hangzhou'}
# 创建API请求对象
request = open_api_models.OpenApiRequest(
  query=OpenApiUtilClient.query(query),
)

# 场景二：设置body参数，reqBodyType的值为json
# body = {
#  'param1': 'value1'
# }
# 创建API请求对象
# request = open_api_models.OpenApiRequest(
#     body=OpenApiUtilClient.query(body),
# )

# 场景三：使用Stream流参数传递文件流
# 创建API请求对象
# request = open_api_models.OpenApiRequest(
#     stream='<FILE_STREAM>', # <FILE_STREAM>需替换为实际文件流
# )

# 场景四：设置body参数，reqBodyType的值为formData
# form_data = {
#  'param1': 'value1'
# }
# 创建API请求对象
# request = open_api_models.OpenApiRequest(
#     body=form_data,
# )

4. 发起请求

通过步骤1创建的client调用call_api发起同步请求，或调用call_api_async发起异步请求。同时，支持为当前请求设置运行时参数，例如超时配置、代理配置等，更多信息请查看进阶配置。

from alibabacloud_tea_util import models as util_models

# 运行时配置，仅对当前请求生效
runtime = util_models.RuntimeOptions(
  # ignore_ssl=True  # 忽略对 SSL 证书的验证，默认验证
  # autoretry=True,  # 是否开启重试 默认关闭 
  # max_attempts=3  # 重试次数 默认3次
)
# 同步调用
response = client.call_api(params, request, runtime)
# 异步调用
# response = await client.call_api_async(params, request, runtime)

# 返回值为 dict 类型，可从返回结果中获得三类数据：body、headers、statusCode（HTTP返回的状态码 ）。
print(response['body'])

代码示例

示例：调用RPC风格的API

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

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
from alibabacloud_openapi_util.client import Client as OpenApiUtilClient


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='DescribeInstanceTypeFamilies',  # API 名称
      method='POST',  # 请求方法
      pathname='/',  # 接口 PATH，RPC接口默认"/"
      protocol='HTTPS',  # 接口协议,
      auth_type='AK',
      req_body_type='json',  # 接口请求体内容格式,
      body_type='json'  # 接口响应体内容格式,
    )

    query = {'RegionId': 'cn-hangzhou'}
    # 创建API请求对象
    request = open_api_models.OpenApiRequest(
      query=OpenApiUtilClient.query(query),
    )
    # 创建运行时选项对象
    runtime = util_models.RuntimeOptions()
    # 发起API调用并打印响应
    response = client.call_api(params, request, runtime)
    # 返回值为Map类型，可从Map中获得三类数据：body、headers、statusCode（HTTP返回的状态码 ）。
    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='DescribeInstanceTypeFamilies',  # API 名称
      method='POST',  # 请求方法
      pathname='/',  # 接口 PATH
      protocol='HTTPS',  # 接口协议,
      auth_type='AK',
      req_body_type='json',  # 接口请求体内容格式,
      body_type='json'  # 接口响应体内容格式,
    )

    query = {'RegionId': 'cn-hangzhou'}
    # 创建API请求对象
    request = open_api_models.OpenApiRequest(
      query=OpenApiUtilClient.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,ROA接口从OpenAPI元数据中data.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()