This topic describes how to use SDK for Python.

Prerequisites

  1. You must have an Alibaba Cloud account and an AccessKey pair (AccessKey ID and AccessKey secret) to use Alibaba Cloud SDK for Python. You can log on to the Security Management page to create or view the AccessKey pair. You can also contact your system administrator to obtain the AccessKey pair.
  2. To use Alibaba Cloud SDK to access API operations of a service, you must activate this service in the Alibaba Cloud Management Console.

Installation

We recommend that you run the following pip commands to install SDK for Python:

pip install aliyun-python-sdk-core # Install the Alibaba Cloud SDK core library.
pip install aliyun-python-sdk-fnf # Install Serverless Workflow SDK.          

Quick start

The following example shows how to use SDK for Python to call Serverless workflow. In this example, a flow is created, an execution is initiated, and the execution details are queried.

Request method

# encoding: utf-8
import time

from aliyunsdkcore.client import AcsClient
from aliyunsdkcore.acs_exception.exceptions import ServerException
from aliyunsdkfnf.request.v20190315 import CreateFlowRequest
from aliyunsdkfnf.request.v20190315 import StartExecutionRequest
from aliyunsdkfnf.request.v20190315 import DescribeExecutionRequest
from aliyunsdkfnf.request.v20190315 import GetExecutionHistoryRequest

flow_definition_type = "FDL"
flow_name = "xxx"
flow_definition = "xxx"
flow_description = "some descriptions"
role_arn = "acs:ram::${Your_Account_ID}:${Your_Role}"
execution_name = "xxx"


def create_flow(fnf_cli):
    request = CreateFlowRequest.CreateFlowRequest()
    request.set_Type(flow_definition_type)
    request.set_Description(flow_description)
    request.set_Definition(flow_definition)
    request.set_RoleArn(role_arn)
    request.set_Name(flow_name)
    return fnf_cli.do_action_with_exception(request)


def start_execution(fnf_cli):
    request = StartExecutionRequest.StartExecutionRequest()
    request.set_FlowName(flow_name)
    request.set_ExecutionName(execution_name)
    return fnf_cli.do_action_with_exception(request)


def describe_execution(fnf_cli):
    request = DescribeExecutionRequest.DescribeExecutionRequest()
    request.set_FlowName(flow_name)
    request.set_ExecutionName(execution_name)
    return fnf_cli.do_action_with_exception(request)


def get_execution_history(fnf_cli):
    request = GetExecutionHistoryRequest.GetExecutionHistoryRequest()
    request.set_FlowName(flow_name)
    request.set_ExecutionName(execution_name)
    return fnf_cli.do_action_with_exception(request)  

Create a client and use the preceding functions to initiate a series of calls.

Note If you need to implement debugging without modification, place the following function in the same file as the preceding request method code block to avoid errors upon import.
def main():

    # Create an AcsClient instance.
    client = AcsClient(
        "<your-access-key-id>",
        "<your-access-key-secret>",
        "<your-region-id>"
    )
    try:
        create_resp = create_flow(client)
        print(create_resp)

        start_resp = start_execution(client)
        print(start_resp)

        time.sleep(1)

        desc_resp = describe_execution(client)
        print(desc_resp)

        get_resp = get_execution_history(client)
        print(get_resp)
    except ServerException as e:
        print(e.get_request_id())


if __name__ == '__main__':
    main()