CloudFlow:Install CloudFlow SDK for Python

Prerequisites

An AccessKey pair is configured. Make sure that an AccessKey pair is created. For more information, see Create an AccessKey pair. To prevent credential leaks, you can write the credentials into environment variables. For more information, see Best practices for using an access credential to call API operations.

Environment requirements

Python 3.7 or later is used.

Step 1: Import CloudFlow SDK for Python to your project

Alibaba Cloud SDK allows you to initiate generic and specialized calls. For more information, see Generic calls and specialized calls. The SDK that you need to import varies based on the call type.

Specialized calls

You can visit OpenAPI Explorer, find the service that you want to use, and view the supported languages and installation methods. Then, import the SDK of the service to your project. In this example, CloudFlow SDK for Python is imported. Perform the following steps to import the SDK:

1. Go to the CloudFlow SDK page in OpenAPI Explorer.

2. In the All languages field, select Python.

3. In the Installation Method section, select an installation method and then copy the installation code to your project.

4. Add the dependency to your project.

Run the following command to install CloudFlow SDK for Python:

pip install alibabacloud_fnf20190315==1.1.3

Generic calls

You do not need to install a service-specific SDK to initiate generic calls. You need to only import the com.aliyun.tea-openapi core package. For information about the latest version of the core package, see tea-openapi. Run the following command to install the core package:

pip install alibabacloud-tea-openapi

Step 2: Initialize a client

Specify the endpoint of CloudFlow based on the region in which you want to use CloudFlow. For information about the endpoint for each region, see Regions.

The following section provides sample code on how to initialize a client for specialized calls. For information about how to initiate generic calls, see Generic calls and specialized calls.

Use an AccessKey pair

# -*- 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:
    """
    Use your AccessKey ID and AccessKey secret to initialize the client.
    @return: Client
    @throws Exception
    """
    config = open_api_models.Config(
    # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured. ,
    access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured. ,
    access_key_secret=os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
    config.endpoint = f'cn-hangzhou.fnf.aliyuncs.com'
    return fnf20190315Client(config) 

Step 3: Use the initialized client to call CloudFlow API operations

API operation: CreateFlow

Call the CreateFlow operation to create a workflow. When you call the CreateFlow operation, create a request and configure parameters and runtime settings based on your business requirements. You can also customize runtime settings to meet specific requirements.

// Create a request.
 create_flow_request = fnf_20190315_models.CreateFlowRequest(
            // Specify a name for the workflow that you want to create.
            name='your_flow_name',
            // Define the workflow against the Flow Definition Language (FDL) syntax standard. To ensure forward compatibility, the system supports the old and new versions of workflow definition specifications. 
            definition='''The old version:
"
type: flow
version: v1
name: my_flow_name
steps:
  - type: pass
    name: mypass
"

The new version:
"
Type: StateMachine
SpecVersion: v1
Name: my_flow_name
StartAt: my_state
States:
  - Type: Pass
    Name: my_state
    End: true
"''',
            // Specify a description for the workflow.  
            description='my test flow',
            // Configure the type of the workflow.
            type='FDL'
        )
        // Configure runtime settings.
        runtime = util_models.RuntimeOptions()

The following sample code provides a complete example on how to create a workflow by using the AccessKey pair to call the CreateFlow operation:

# -*- 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:
        """
        Use your AccessKey ID and AccessKey secret to initialize the client.
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured. ,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured. ,
            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='''The old version:
"
type: flow
version: v1
name: my_flow_name
steps:
  - type: pass
    name: mypass
"

The new version:
"
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:
            # If you copy and run the sample code, write your own code to display the response of the API operation.
            client.create_flow_with_options(create_flow_request, runtime)
        except Exception as error:
            # Handle exceptions with caution based on your actual business scenario and do not ignore exceptions in your project. The error messages displayed in this example are for reference only. 
            # Display error messages.
            print(error.message)
            # Provide the URL for troubleshooting.
            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 a request.
 create_flow_request = fnf_20190315_models.CreateFlowRequest(
            // Specify a name for the workflow that you want to create.
            name='your_flow_name',
            // Define the workflow against the FDL syntax standard. To ensure forward compatibility, the system supports the old and new versions of workflow definition specifications. 
            definition='''The old version:
"
type: flow
version: v1
name: my_flow_name
steps:
  - type: pass
    name: mypass
"

The new version:
"
Type: StateMachine
SpecVersion: v1
Name: my_flow_name
StartAt: my_state
States:
  - Type: Pass
    Name: my_state
    End: true
"''',
            // Specify a description for the workflow.  
            description='my test flow',
            // Configure the type of the workflow.
            type='FDL'
        )
        // Configure runtime settings.
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy and run the sample code, write your own code to display the response of the API operation.
            await client.create_flow_with_options_async(create_flow_request, runtime)
        except Exception as error:
            # Handle exceptions with caution based on your actual business scenario and do not ignore exceptions in your project. The error messages displayed in this example are for reference only. 
            # Display error messages.
            print(error.message)
            # Provide the URL for troubleshooting.
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

API operation: DescribeFlow

Call the DescribeFlow operation to query the information of a workflow. When you call the DescribeFlow operation, create a request and configure parameters and runtime settings based on your business requirements. You can also customize runtime settings to meet specific requirements.

// Create a request.
describe_flow_request = fnf_20190315_models.DescribeFlowRequest(
            // Specify the name of the workflow that you want to query.
            name='your_flow_name'
        )
        // Configure runtime settings.
        runtime = util_models.RuntimeOptions() 

The following sample code provides a complete example on how to query the information of a workflow by using the AccessKey pair to call the DescribeFlow operation:

# -*- 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:
        """
        Use your AccessKey ID and AccessKey secret to initialize the client.
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured. ,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured. ,
            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(
            // Specify the name of the workflow that you want to query.
            name='your_flow_name'
        )
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy and run the sample code, write your own code to display the response of the API operation.
            client.describe_flow_with_options(describe_flow_request, runtime)
        except Exception as error:
            # Handle exceptions with caution based on your actual business scenario and do not ignore exceptions in your project. The error messages displayed in this example are for reference only. 
            # Display error messages.
            print(error.message)
            # Provide the URL for troubleshooting.
            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:
            # If you copy and run the sample code, write your own code to display the response of the API operation.
            await client.describe_flow_with_options_async(describe_flow_request, runtime)
        except Exception as error:
            # Handle exceptions with caution based on your actual business scenario and do not ignore exceptions in your project. The error messages displayed in this example are for reference only. 
            # Display error messages.
            print(error.message)
            # Provide the URL for troubleshooting.
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:]) 

API operation: StartExecution

The StartExecution operation is an asynchronous operation that starts the execution of a workflow. When you call the StartExecution operation, create a request and configure parameters and runtime settings based on your business requirements. You can also customize runtime settings to meet specific requirements.

// Create a request.
 start_execution_request = fnf_20190315_models.StartExecutionRequest(
            // Specify the name of the workflow that you want to execute.
            flow_name='your_flow_name',
            // Specify a name for the execution.
            execution_name='your_exec_name',
            // Specify the input of the execution.
            input='{"key":"value"}',
            // Call back TaskToken-related tasks after the workflow execution ends. Example value: 12.
            callback_fn_ftask_token='12'
        )
        // Configure runtime settings.
        runtime = util_models.RuntimeOptions()

The following sample code provides a complete example on how to start the execution of a workflow by using the AccessKey pair to call the StartExecution operation:

# -*- 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:
        """
        Use your AccessKey ID and AccessKey secret to initialize the client.
        @return: Client
        @throws Exception
        """
        config = open_api_models.Config(
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_ID environment variable is configured. ,
            access_key_id=os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
            # Required. Make sure that the ALIBABA_CLOUD_ACCESS_KEY_SECRET environment variable is configured. ,
            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:
            # If you copy and run the sample code, write your own code to display the response of the API operation.
            client.start_execution_with_options(start_execution_request, runtime)
        except Exception as error:
            # Handle exceptions with caution based on your actual business scenario and do not ignore exceptions in your project. The error messages displayed in this example are for reference only. 
            # Display error messages.
            print(error.message)
            # Provide the URL for troubleshooting.
            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 a request.
        start_execution_request = fnf_20190315_models.StartExecutionRequest(
            // Specify the name of the workflow that you want to execute.
            flow_name='your_flow_name',
            // Specify a name for the execution.
            execution_name='your_exec_name',
            // Specify the input of the execution.
            input='{"key":"value"}',
            // Call back TaskToken-related tasks after the workflow execution ends. Example value: 12.
            callback_fn_ftask_token='12'
        )
        // Configure runtime settings.
        runtime = util_models.RuntimeOptions()
        try:
            # If you copy and run the sample code, write your own code to display the response of the API operation.
            await client.start_execution_with_options_async(start_execution_request, runtime)
        except Exception as error:
            # Handle exceptions with caution based on your actual business scenario and do not ignore exceptions in your project. The error messages displayed in this example are for reference only. 
            # Display error messages.
            print(error.message)
            # Provide the URL for troubleshooting.
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


if __name__ == '__main__':
    Sample.main(sys.argv[1:])

SDK call example

You can use the API-level SDK demos in different programming languages for debugging. To view the sample code, visit OpenAPI Explorer.