Python 用 CloudFlow SDK のインストールと使用 - CloudFlow - Alibaba Cloud ドキュメントセンター

このトピックでは、CloudFlow SDK for Python のインストール方法と、SDK を使用して CloudFlow API オペレーションを呼び出す方法の例を示します。以下の例では、Python 用 CloudFlow SDK を使用して、CloudFlow API オペレーションを呼び出し、ワークフローの作成、ワークフローに関する情報のクエリ、ワークフローの実行の非同期開始を行います。このトピックでは、完全な SDK 統合手順についても説明します。

前提条件

AccessKey ペアが構成されていること。 AccessKey ペアが作成されていることを確認してください。詳細については、「AccessKey ペアの作成」をご参照ください。資格情報の漏洩を防ぐために、資格情報を環境変数に書き込むことができます。詳細については、「アクセス資格情報を使用して API オペレーションを呼び出すためのベストプラクティス」をご参照ください。

環境要件

Python 3.7 以後を使用します。

手順 1: プロジェクトに Python 用 CloudFlow SDK をインポートする

Alibaba Cloud SDK を使用すると、汎用呼び出しと専用呼び出しを開始できます。詳細については、「汎用呼び出しと専用呼び出し」をご参照ください。インポートする必要のある SDK は、呼び出しの種類によって異なります。

専用呼び出し

OpenAPI Explorer にアクセスし、使用するサービスを見つけて、サポートされている言語とインストール方法を確認できます。次に、サービスの SDK をプロジェクトにインポートします。この例では、Python 用 CloudFlow SDK をインポートします。 SDK をインポートするには、次の手順を実行します。

OpenAPI Explorer の CloudFlow SDK ページに移動します。
[すべての言語] フィールドで、[Python] を選択します。
[インストール方法] セクションで、インストール方法を選択し、インストールコードをプロジェクトにコピーします。
依存関係をプロジェクトに追加します。

Python 用 CloudFlow SDK をインストールするには、次のコマンドを実行します。

pip install alibabacloud_fnf20190315==1.1.3

汎用呼び出し

汎用呼び出しを開始するために、サービス固有の SDK をインストールする必要はありません。 com.aliyun.tea-openapi コアパッケージのみをインポートする必要があります。コアパッケージの最新バージョンについては、「tea-openapi」をご参照ください。コアパッケージをインストールするには、次のコマンドを実行します。

pip install alibabacloud-tea-openapi

手順 2: クライアントを初期化する

CloudFlow を使用するリージョンに基づいて、CloudFlow のエンドポイントを指定します。各リージョンのエンドポイントについては、「リージョン」をご参照ください。

次のセクションでは、専用呼び出しのクライアントを初期化する方法のサンプルコードを示します。汎用呼び出しを開始する方法については、「汎用呼び出しと専用呼び出し」をご参照ください。

AccessKey ペアを使用する

説明

Alibaba Cloud アカウントの AccessKey ペアには、すべての API オペレーションに対する権限があります。 AccessKey ペアを使用して操作を実行することは、リスクの高い操作です。 API オペレーションの呼び出しや日常の O&M の実行には、Resource Access Management (RAM) ユーザーを使用することをお勧めします。 AccessKey ID と AccessKey シークレットをプロジェクトコードに保存しないことをお勧めします。そうしないと、AccessKey ペアが漏洩し、アカウント内のすべてのリソースのセキュリティが侵害される可能性があります。

この例では、ALIBABA_CLOUD_ACCESS_KEY_ID 環境変数と ALIBABA_CLOUD_ACCESS_KEY_SECRET 環境変数から AccessKey ペアを取得して、ID 認証を実装します。

認証情報の構成方法の詳細については、「アクセス資格情報の管理」をご参照ください。
環境変数を構成する方法は、オペレーティングシステムによって異なります。詳細については、「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:
    """
    AccessKey ID と AccessKey シークレットを使用してクライアントを初期化します。
    @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)

手順 3: 初期化されたクライアントを使用して CloudFlow API オペレーションを呼び出す

説明

クライアントを初期化したら、クライアントを使用して CloudFlow API オペレーションを呼び出すことができます。

API オペレーション: CreateFlow

CreateFlow オペレーションを呼び出して、ワークフローを作成します。 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()

次のサンプルコードは、AccessKey ペアを使用して CreateFlow オペレーションを呼び出すことによってワークフローを作成する方法の完全な例を示しています。

# -*- 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:
        """
        AccessKey ID と AccessKey シークレットを使用してクライアントを初期化します。
        @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:
            # 実際のビジネスシナリオに基づいて例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例に表示されているエラーメッセージは参照用です。
            # エラーメッセージを表示します。
            print(error.message)
            # トラブルシューティング用の URL を提供します。
            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',
            // 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:
            # 実際のビジネスシナリオに基づいて例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例に表示されているエラーメッセージは参照用です。
            # エラーメッセージを表示します。
            print(error.message)
            # トラブルシューティング用の URL を提供します。
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


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

API オペレーション: DescribeFlow

DescribeFlow オペレーションを呼び出して、ワークフローの情報をクエリします。 DescribeFlow オペレーションを呼び出すときは、ビジネス要件に基づいてリクエストを作成し、パラメーターとランタイム設定を構成します。特定の要件を満たすために、ランタイム設定をカスタマイズすることもできます。

// リクエストを作成します。
describe_flow_request = fnf_20190315_models.DescribeFlowRequest(
            // クエリするワークフローの名前を指定します。
            name='your_flow_name'
        )
        // ランタイム設定を構成します。
        runtime = util_models.RuntimeOptions()

次のサンプルコードは、AccessKey ペアを使用して DescribeFlow オペレーションを呼び出すことによってワークフローの情報をクエリする方法の完全な例を示しています。

# -*- 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:
        """
        AccessKey ID と AccessKey シークレットを使用してクライアントを初期化します。
        @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:
            # 実際のビジネスシナリオに基づいて例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例に表示されているエラーメッセージは参照用です。
            # エラーメッセージを表示します。
            print(error.message)
            # トラブルシューティング用の URL を提供します。
            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:
            # 実際のビジネスシナリオに基づいて例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例に表示されているエラーメッセージは参照用です。
            # エラーメッセージを表示します。
            print(error.message)
            # トラブルシューティング用の URL を提供します。
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


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

API オペレーション: StartExecution

StartExecution オペレーションは、ワークフローの実行を開始する非同期オペレーションです。 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()

次のサンプルコードは、AccessKey ペアを使用して StartExecution オペレーションを呼び出すことによってワークフローの実行を開始する方法の完全な例を示しています。

# -*- 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:
        """
        AccessKey ID と AccessKey シークレットを使用してクライアントを初期化します。
        @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:
            # 実際のビジネスシナリオに基づいて例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例に表示されているエラーメッセージは参照用です。
            # エラーメッセージを表示します。
            print(error.message)
            # トラブルシューティング用の URL を提供します。
            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:
            # 実際のビジネスシナリオに基づいて例外を慎重に処理し、プロジェクトで例外を無視しないでください。この例に表示されているエラーメッセージは参照用です。
            # エラーメッセージを表示します。
            print(error.message)
            # トラブルシューティング用の URL を提供します。
            print(error.data.get("Recommend"))
            UtilClient.assert_as_string(error.message)


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

SDK 呼び出し例

デバッグには、さまざまなプログラミング言語の API レベルの SDK デモを使用できます。サンプルコードを表示するには、OpenAPI Explorer にアクセスしてください。