CloudFlow:Install CloudFlow SDK for Python

Specialized calls (recommended)

Run the following command to install the CloudFlow SDK:

pip install alibabacloud_fnf20190315==1.1.3

To verify the installation:

python -c "import alibabacloud_fnf20190315; print('CloudFlow SDK installed successfully')"

You can find the latest version and alternative installation methods on the CloudFlow SDK page in OpenAPI Explorer. Select Python in the All languages field.

Generic calls

Generic calls require only the core alibabacloud-tea-openapi package -- no product-specific SDK needed:

pip install alibabacloud-tea-openapi

For the latest core package version, see tea-openapi.

CreateFlow

CreateFlow creates a flow. Provide a name, a Flow Definition Language (FDL) definition, and a type.

CloudFlow supports two FDL specification versions. The old version uses type: flow, and the new version uses Type: StateMachine. Both are accepted for forward compatibility.

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:
        """Initialize a CloudFlow client with AccessKey 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 = "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="<your-flow-name>",
            definition="""\
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:
            response = client.create_flow_with_options(create_flow_request, runtime)
            print(response.body)
        except Exception as error:
            print(error.message)
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


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

For asynchronous calls, replace create_flow_with_options with create_flow_with_options_async and use await.

Replace the following placeholders with your values:

DescribeFlow

DescribeFlow retrieves the details of a flow, including its definition and metadata.

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:
        """Initialize a CloudFlow client with AccessKey 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 = "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:
            response = client.describe_flow_with_options(describe_flow_request, runtime)
            print(response.body)
        except Exception as error:
            print(error.message)
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


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

Replace <your-flow-name> with the name of the flow to query. For asynchronous calls, use describe_flow_with_options_async with await.

StartExecution

StartExecution starts an asynchronous execution of a flow.

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:
        """Initialize a CloudFlow client with AccessKey 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 = "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-execution-name>",
            input='{"key":"value"}',
            callback_fn_ftask_token="12",
        )
        runtime = util_models.RuntimeOptions()

        try:
            response = client.start_execution_with_options(start_execution_request, runtime)
            print(response.body)
        except Exception as error:
            print(error.message)
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


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

Replace the following placeholders with your values:

For asynchronous calls, use start_execution_with_options_async with await.