泛化调用
阿里云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 | 请求参数显示 |
body | 请求参数显示 |
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()