Python V2.0 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

以調用Container Service查詢叢集列表資訊為例，展示如何使用泛化調用。

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()