Google Cloud WorkflowsをPythonから引数指定で実行してみる


# Workflow登録

引数を取って、30秒スリープ後、結果出力という簡単なワークフローを登録します。


```workflow.yaml
main:
  params: [args]
  steps:
    - map_params:
        assign:
          - outputVar: ${"Hello " + args.firstName + " " + args.lastName}
    - sleep:
        call: sys.sleep
        args:
          seconds: 30
    - output:
        return: ${outputVar}
```



```sh
$ gcloud beta workflows deploy myworkflow --location us-central1 --source workflow.yaml 
```

以下の用に登録されます。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/116068/00b20893-3b53-e7cf-63c2-7c5a84584f69.png)

# PythonによるWorkflow起動 ：非同期

Workflowに名前を引数として渡すコードを作ってみます。

```run_workflow.py
import json
from google.cloud.workflows import WorkflowsClient
from google.cloud.workflows.executions import (
    ExecutionsClient,
    Execution,
)

GCP_PROJECT = "your-googlecloud-project"
WF_LOCATION = "us-central1"
WF_NAME = "myworkflow"


WF_ARGS = {"firstName": "Yusuke", "lastName": "Otomo"}

if __name__ == "__main__":
    workflow_name = WorkflowsClient().workflow_path(
        project=GCP_PROJECT, location=WF_LOCATION, workflow=WF_NAME
    )  # workflow_name -> projects/your-gcp-project/locations/us-central1/workflows/myworkflow

    cli = ExecutionsClient()
    cli.create_execution(
        parent=workflow_name,
        execution=Execution(argument=json.dumps(WF_ARGS)),
    )
```

実行します。

```sh
$ python run_workflow.py
```

非同期なので、スクリプトはWorkflowをトリガーしてすぐに返ります。

Google Cloudの画面で見るとワークフローが実行中であることがわかります。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/116068/fc2611c3-ef05-5c1c-92a9-32695211a7c8.png)

想定通り引数も渡っています。

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/116068/5b77d094-dae2-1864-b50e-6dd9aff81ce3.png)

# PythonによるWorkflow起動 ：同期
Workflowの結果を取得するため、ステータスをポーリングしてみます

```run_workflow_sync.py
import json
from time import sleep
from google.cloud.workflows import WorkflowsClient
from google.cloud.workflows.executions import (
    ExecutionsClient,
    Execution,
    GetExecutionRequest,
)

GCP_PROJECT = "your-googlecloud-project"
WF_LOCATION = "us-central1"
WF_NAME = "myworkflow"


WF_ARGS = {"firstName": "Yusuke", "lastName": "Otomo"}

if __name__ == "__main__":
    workflow_name = WorkflowsClient().workflow_path(
        project=GCP_PROJECT, location=WF_LOCATION, workflow=WF_NAME
    )  # workflow_name -> projects/your-gcp-project/locations/us-central1/workflows/myworkflow

    cli = ExecutionsClient()
    create_response = cli.create_execution(
        parent=workflow_name,
        execution=Execution(argument=json.dumps(WF_ARGS)),
    )

    while True:
        get_response = cli.get_execution(
            request=GetExecutionRequest(name=create_response.name)
        )
        if get_response.state != Execution.State.ACTIVE:
            print(get_response)
            break

        print(f"waiting... State ->{get_response.state}")
        sleep(5)
```

実行すると以下のようにWorkflowのOutputの結果を取得できました。

```sh
$ python run_workflow_sync.py

waiting... State -> ACTIVE
waiting... State -> ACTIVE
waiting... State -> ACTIVE
waiting... State -> ACTIVE
waiting... State -> ACTIVE
waiting... State -> ACTIVE
Workflow Result -> "Hello Yusuke Otomo"
```

# 参考

https://github.com/googleapis/python-workflows