本文將詳細介紹云工作流的Python SDK安裝步驟及調用示例。以Python
語言為例,本文將講解如何調用云工作流以創建流程接口、獲取流程相關信息接口,以及異步啟動流程執行的接口為例,并提供完整的集成步驟。
前提條件
調用阿里云OpenAPI通常需要設置訪問密鑰(AccessKey)。請確保已創建AccessKey。具體操作,請參見創建AccessKey。為了避免憑據泄露,常見的方案是將其寫入到環境變量中,更多安全方案請參見使用訪問憑據訪問阿里云OpenAPI最佳實踐。
環境要求
Python版本 >= 3.7。
步驟一:引入SDK
阿里云SDK支持泛化與特化兩種方式調用OpenAPI,詳情參見泛化調用與特化調用,不同的調用方式需要引入的SDK也不同。
特化調用
您可以訪問OpenAPI門戶網站,搜索您需要的產品,查看產品支持的SDK語言及安裝方法,然后在您的項目中引入SDK。本案例的SDK獲取步驟如下:
訪問云工作流 SDK。
在所有語言欄目中選擇您需要的SDK語言。
選擇您需要的安裝方式,將代碼復制到您的項目中。
在您的項目中載入該依賴包。
其Python語言安裝產品如下:
pip install alibabacloud_fnf20190315==1.1.3
泛化調用
泛化調用方式不依賴任何一個產品的SDK,只依賴如下核心包com.aliyun.tea-openapi
。其Python語言安裝配置文件如下,最新版本請參見tea-openapi。
pip install alibabacloud-tea-openapi
步驟二:初始化Client
請根據云工作流(FNF)所屬地域正確填寫服務接入地址(又稱“訪問端點”或“Endpoint”),關于服務接入地址的更多信息,請參見支持的地域。
以下將以特化調用代碼為例,詳細說明調用過程。如您選擇泛化調用方案,更多信息請參見泛化調用與特化調用。
使用AK初始化
阿里云賬號AccessKey擁有所有OpenAPI的訪問權限,建議您使用RAM用戶進行API訪問或日常運維。強烈建議不要把AccessKey ID和AccessKey Secret保存到工程代碼里,否則可能導致AccessKey泄露,威脅您賬號下所有資源的安全。
本示例以將AccessKey配置在環境變量ALIBABA_CLOUD_ACCESS_KEY_ID和ALIBABA_CLOUD_ACCESS_KEY_SECRET的方式來實現身份驗證為例。
更多認證信息配置方式,請參見管理訪問憑據。
不同操作系統的環境變量配置方法不同,具體操作,請參見在Linux、macOS和Windows系統配置環境變量。
# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_fnf20190315.client import Client as fnf20190315Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_fnf20190315 import models as fnf_20190315_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
@staticmethod
def create_client() -> fnf20190315Client:
"""
使用AK&SK初始化賬號Client
@return: Client
@throws Exception
"""
config = open_api_models.Config(
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
config.endpoint = f'cn-hangzhou.fnf.aliyuncs.com'
return fnf20190315Client(config)
步驟三:使用已初始化的Client調用云工作流 API
初始化Client后,您可以通過Client調用云工作流API。
接口名稱:CreateFlow
此接口用于創建一個流程。在調用過程中,您需要根據實際業務需求創建請求對象,并設置相應的參數及運行時配置。同時,您也可以自定義運行時配置以滿足特定需求
// 創建請求對象
create_flow_request = fnf_20190315_models.CreateFlowRequest(
// 您的流程名稱
name='your_flow_name',
// 流程定義,遵循 Flow Definition Language (FDL)語法標準??紤]到向前兼容,當系統支持兩種規范的流程定義規范。
definition='''舊版:
"
type: flow
version: v1
name: my_flow_name
steps:
- type: pass
name: mypass
"
新版:
"
Type: StateMachine
SpecVersion: v1
Name: my_flow_name
StartAt: my_state
States:
- Type: Pass
Name: my_state
End: true
"''',
// 流程描述
description='my test flow',
// 流程類型
type='FDL'
)
// 運行時配置
runtime = util_models.RuntimeOptions()
以下是使用AK創建一個流程的完整示例代碼:
# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_fnf20190315.client import Client as fnf20190315Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_fnf20190315 import models as fnf_20190315_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
def __init__(self):
pass
@staticmethod
def create_client() -> fnf20190315Client:
"""
使用AK&SK初始化賬號Client
@return: Client
@throws Exception
"""
config = open_api_models.Config(
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
config.endpoint = f'cn-hangzhou.fnf.aliyuncs.com'
return fnf20190315Client(config)
@staticmethod
def main(
args: List[str],
) -> None:
client = Sample.create_client()
create_flow_request = fnf_20190315_models.CreateFlowRequest(
name='my_flow_name',
definition='''舊版:
"
type: flow
version: v1
name: my_flow_name
steps:
- type: pass
name: mypass
"
新版:
"
Type: StateMachine
SpecVersion: v1
Name: my_flow_name
StartAt: my_state
States:
- Type: Pass
Name: my_state
End: true
"''',
description='my test flow',
type='FDL'
)
runtime = util_models.RuntimeOptions()
try:
# 復制代碼運行請自行打印 API 的返回值
client.create_flow_with_options(create_flow_request, runtime)
except Exception as error:
# 此處僅做打印展示,請謹慎對待異常處理,在工程項目中切勿直接忽略異常。
# 錯誤 message
print(error.message)
# 診斷地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
@staticmethod
async def main_async(
args: List[str],
) -> None:
client = Sample.create_client()
// 創建請求對象
create_flow_request = fnf_20190315_models.CreateFlowRequest(
// 您的流程名稱
name='your_flow_name',
// 流程定義,遵循 Flow Definition Language (FDL)語法標準??紤]到向前兼容,當系統支持兩種規范的流程定義規范。
definition='''舊版:
"
type: flow
version: v1
name: my_flow_name
steps:
- type: pass
name: mypass
"
新版:
"
Type: StateMachine
SpecVersion: v1
Name: my_flow_name
StartAt: my_state
States:
- Type: Pass
Name: my_state
End: true
"''',
// 流程描述
description='my test flow',
// 流程類型
type='FDL'
)
// 運行時配置
runtime = util_models.RuntimeOptions()
try:
# 復制代碼運行請自行打印 API 的返回值
await client.create_flow_with_options_async(create_flow_request, runtime)
except Exception as error:
# 此處僅做打印展示,請謹慎對待異常處理,在工程項目中切勿直接忽略異常。
# 錯誤 message
print(error.message)
# 診斷地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
Sample.main(sys.argv[1:])
接口名稱:DescribeFlow
此接口用于獲取特定流程的相關信息。在調用過程中,您需要根據實際業務需求創建請求對象,并設置相應的參數及運行時配置。同時,您也可以自定義運行時配置以滿足特定需求。
// 創建請求對象
describe_flow_request = fnf_20190315_models.DescribeFlowRequest(
// 您的流程名稱
name='your_flow_name'
)
// 運行時配置
runtime = util_models.RuntimeOptions()
以下是使用AK獲取流程相關信息的完整示例代碼:
# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_fnf20190315.client import Client as fnf20190315Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_fnf20190315 import models as fnf_20190315_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
def __init__(self):
pass
@staticmethod
def create_client() -> fnf20190315Client:
"""
使用AK&SK初始化賬號Client
@return: Client
@throws Exception
"""
config = open_api_models.Config(
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
config.endpoint = f'cn-hangzhou.fnf.aliyuncs.com'
return fnf20190315Client(config)
@staticmethod
def main(
args: List[str],
) -> None:
client = Sample.create_client()
describe_flow_request = fnf_20190315_models.DescribeFlowRequest(
// 您的流程名稱
name='your_flow_name'
)
runtime = util_models.RuntimeOptions()
try:
# 復制代碼運行請自行打印 API 的返回值
client.describe_flow_with_options(describe_flow_request, runtime)
except Exception as error:
# 此處僅做打印展示,請謹慎對待異常處理,在工程項目中切勿直接忽略異常。
# 錯誤 message
print(error.message)
# 診斷地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
@staticmethod
async def main_async(
args: List[str],
) -> None:
client = Sample.create_client()
describe_flow_request = fnf_20190315_models.DescribeFlowRequest(
name='my_flow_name'
)
runtime = util_models.RuntimeOptions()
try:
# 復制代碼運行請自行打印 API 的返回值
await client.describe_flow_with_options_async(describe_flow_request, runtime)
except Exception as error:
# 此處僅做打印展示,請謹慎對待異常處理,在工程項目中切勿直接忽略異常。
# 錯誤 message
print(error.message)
# 診斷地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
Sample.main(sys.argv[1:])
接口名稱:StartExecution
此接口用于異步調用以啟動一個流程的執行。在調用過程中,您需要根據實際業務需求構建請求對象,并設置相應的參數及運行時配置。同時,您也可以自定義運行時配置以滿足特定需求。
// 創建請求對象
start_execution_request = fnf_20190315_models.StartExecutionRequest(
// 您的開始執行流程名稱
flow_name='your_flow_name',
// 執行名稱
execution_name='your_exec_name',
// 執行輸入
input='{"key":"value"}',
// 流程執行結束后回調TaskToken相關任務 示例值:12
callback_fn_ftask_token='12'
)
// 運行時配置
runtime = util_models.RuntimeOptions()
以下是使用AK異步調用啟動一個流程的執行的完整示例代碼:
# -*- coding: utf-8 -*-
import os
import sys
from typing import List
from alibabacloud_fnf20190315.client import Client as fnf20190315Client
from alibabacloud_tea_openapi import models as open_api_models
from alibabacloud_fnf20190315 import models as fnf_20190315_models
from alibabacloud_tea_util import models as util_models
from alibabacloud_tea_util.client import Client as UtilClient
class Sample:
def __init__(self):
pass
@staticmethod
def create_client() -> fnf20190315Client:
"""
使用AK&SK初始化賬號Client
@return: Client
@throws Exception
"""
config = open_api_models.Config(
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_ID。,
access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
# 必填,請確保代碼運行環境設置了環境變量 ALIBABA_CLOUD_ACCESS_KEY_SECRET。,
access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)
config.endpoint = f'cn-hangzhou.fnf.aliyuncs.com'
return fnf20190315Client(config)
@staticmethod
def main(
args: List[str],
) -> None:
client = Sample.create_client()
start_execution_request = fnf_20190315_models.StartExecutionRequest(
flow_name='your_flow_name',
execution_name='your_exec_name',
input='{"key":"value"}',
callback_fn_ftask_token='12'
)
runtime = util_models.RuntimeOptions()
try:
# 復制代碼運行請自行打印 API 的返回值
client.start_execution_with_options(start_execution_request, runtime)
except Exception as error:
# 此處僅做打印展示,請謹慎對待異常處理,在工程項目中切勿直接忽略異常。
# 錯誤 message
print(error.message)
# 診斷地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
@staticmethod
async def main_async(
args: List[str],
) -> None:
client = Sample.create_client()
// 創建請求對象
start_execution_request = fnf_20190315_models.StartExecutionRequest(
// 您的開始執行流程名稱
flow_name='your_flow_name',
// 執行名稱
execution_name='your_exec_name',
// 執行輸入
input='{"key":"value"}',
// 流程執行結束后回調TaskToken相關任務 示例值:12
callback_fn_ftask_token='12'
)
// 運行時配置
runtime = util_models.RuntimeOptions()
try:
# 復制代碼運行請自行打印 API 的返回值
await client.start_execution_with_options_async(start_execution_request, runtime)
except Exception as error:
# 此處僅做打印展示,請謹慎對待異常處理,在工程項目中切勿直接忽略異常。
# 錯誤 message
print(error.message)
# 診斷地址
print(error.data.get("Recommend"))
UtilClient.assert_as_string(error.message)
if __name__ == '__main__':
Sample.main(sys.argv[1:])
SDK調用示例
您可以使用API級別的多語言SDK Demo進行調試。示例代碼,請參見開發者門戶OpenAPI Explorer。