AI 助理
备案控制台登录注册
官方文档
输入文档关键字查找
首页阿里云SDK开发参考Python SDKV2.0 Python SDK（推荐）泛化调用

泛化调用

更新时间：2025-03-21 07:45:57
我的收藏

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

特点

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

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

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

使用说明

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

安装核心SDK

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

pip install alibabacloud-tea-openapi

调用OpenAPI

初始化请求客户端

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

        # 从环境变量中获取访问密钥
        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)


        # 使用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)

配置OpenAPI信息

通过open_api_models.Params配置OpenAPI的基本信息，比如OpenAPI的风格、API版本、请求方式等信息。

  params = open_api_models.Params(
            style='RPC',  # API风格：RPC或ROA。
            version='2014-05-26',  # API版本号
            action='DescribeRegions',  # 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。
        )

配置请求参数

通过open_api_models.OpenApiRequest配置请求参数，请求参数支持通过查询参数（query）和请求体（body）传递。

请求参数传递方式

描述

query

通过查询参数传递参数。

说明

如果RPC接口的请求参数采用query传参方式，则该请求参数同样支持使用body进行传递。

body

通过请求体（body）传递参数。在使用请求体（body）传递参数时，需要根据请求体的类型设置reqBodyType的值。

        # 方式一：设置查询参数（query）
        query = {'InstanceChargeType': 'PrePaid'}
        # 创建API请求对象
        request = open_api_models.OpenApiRequest(
            query=query,
        )

        # # 方式二：设置body参数，reqBodyType的值为json
        # body = {'InstanceChargeType': 'PrePaid'}
        # json_body = json.dumps(body)
        # # 创建API请求对象
        # request = open_api_models.OpenApiRequest(
        #     body=json_body,
        # )

        # # 方式三：设置body参数，reqBodyType的值为byte
        # body = {'InstanceChargeType': 'PrePaid'}
        # bytes_body = json.dumps(body).encode('utf-8')
        # # 创建API请求对象
        # request = open_api_models.OpenApiRequest(
        #     stream=bytes_body,
        # )

        # # 方式四：设置body参数，reqBodyType的值为formData
        # body = {'InstanceChargeType': 'PrePaid'}
        # form_data = urllib.parse.urlencode(body).encode('utf-8')
        # # 创建API请求对象
        # request = open_api_models.OpenApiRequest(
        #     body=form_data,
        # )

发起请求

通过客户端client调用call_api发起同步请求或调用call_api_async发起异步请求。

        # 创建运行时选项对象
        runtime = util_models.RuntimeOptions()
        # 同步调用
        response = client.call_api(params, request, runtime)
        # 异步调用
        # response = await client.call_api_async(params, request, runtime)
        # 返回值为Map类型，可从Map中获得三类数据：body、headers、statusCode（HTTP返回的状态码 ）。
        print(response)

代码示例

示例：调用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，RPC接口默认"/"
            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)
        # 返回值为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='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,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()
上一篇：异常处理下一篇：V1.0 Python SDK
  • 本页导读 （1）
  • 特点
  • 使用说明
  • 安装核心SDK
  • 调用OpenAPI
  • 初始化请求客户端
  • 配置OpenAPI信息
  • 配置请求参数
  • 发起请求
  • 代码示例
  • 示例：调用RPC风格的API
  • 示例：调用RESTful（ROA）风格的API
AI助理

点击开启售前

在线咨询服务

你好，我是AI助理

可以解答问题、推荐解决方案等