Add basic workflow (#8)

Author: lwolczynski

basic/README.md

@@ -0,0 +1,6 @@
+### How to run
+
+* start a iWF server following the [instructions](https://github.com/indeedeng/iwf#how-to-use)
+* build and run `main.py`
+* start a workflow: `http://localhost:8802/basic/start?workflowId=test-1108&inputNum=4`
+* watch in WebUI `http://localhost:8233/namespaces/default/workflows`

basic/basic_workflow.py

@@ -0,0 +1,87 @@
+from iwf.command_request import CommandRequest, InternalChannelCommand, TimerCommand
+from iwf.command_results import CommandResults
+from iwf.communication import Communication
+from iwf.communication_schema import CommunicationSchema, CommunicationMethod
+from iwf.iwf_api.models import ChannelRequestStatus
+from iwf.persistence import Persistence
+from iwf.persistence_schema import PersistenceSchema, PersistenceField
+from iwf.rpc import rpc
+from iwf.state_decision import StateDecision
+from iwf.state_schema import StateSchema
+from iwf.workflow import ObjectWorkflow
+from iwf.workflow_context import WorkflowContext
+from iwf.workflow_state import WorkflowState
+
+TEST_APPROVAL_KEY = "Approval"
+TEST_STRING_KEY = "TestString"
+
+class BasicWorkflow(ObjectWorkflow):
+    def get_workflow_states(self) -> StateSchema:
+        return StateSchema.with_starting_state(
+            BasicWorkflowState1(),
+            BasicWorkflowState2())
+
+
+    def get_persistence_schema(self) -> PersistenceSchema:
+        return PersistenceSchema.create(
+            PersistenceField.data_attribute_def(TEST_STRING_KEY, str),
+        )
+
+    def get_communication_schema(self) -> CommunicationSchema:
+        return CommunicationSchema.create(
+            CommunicationMethod.internal_channel_def(TEST_APPROVAL_KEY, str)
+        )
+
+    @rpc()
+    def append_string(self, st: str, persistence: Persistence) -> str:
+        current = persistence.get_data_attribute(TEST_STRING_KEY)
+        if current is None:
+            current = ""
+        current = current + ", " + st
+        persistence.set_data_attribute(TEST_STRING_KEY, current)
+        return current
+
+    @rpc()
+    def approve(self, communication: Communication):
+        communication.publish_to_internal_channel(TEST_APPROVAL_KEY, "approved")
+
+class BasicWorkflowState1(WorkflowState[int]):
+    def execute(
+            self,
+            ctx: WorkflowContext,
+            data: int,
+            command_results: CommandResults,
+            persistence: Persistence,
+            communication: Communication,
+    ) -> StateDecision:
+        output = data + 1
+        return StateDecision.single_next_state(BasicWorkflowState2, output)
+
+class BasicWorkflowState2(WorkflowState[int]):
+    def wait_until(
+            self,
+            ctx: WorkflowContext,
+            data: int,
+            persistence: Persistence,
+            communication: Communication,
+    ) -> CommandRequest:
+        return CommandRequest.for_any_command_completed(
+            InternalChannelCommand.by_name(TEST_APPROVAL_KEY),
+            TimerCommand.by_seconds(data)
+        )
+
+    def execute(
+            self,
+            ctx: WorkflowContext,
+            data: int,
+            command_results: CommandResults,
+            persistence: Persistence,
+            communication: Communication,
+    ) -> StateDecision:
+        internal_channel_result = command_results.internal_channel_commands[0]
+        if internal_channel_result.status == ChannelRequestStatus.RECEIVED:
+            return StateDecision.graceful_complete_workflow(internal_channel_result.value)
+        else:
+            return StateDecision.single_next_state(BasicWorkflowState2, data)
+
+

basic/iwf_config.py

@@ -0,0 +1,13 @@
+from iwf.client import Client
+from iwf.registry import Registry
+from iwf.worker_service import (
+    WorkerService,
+)
+
+from basic.basic_workflow import BasicWorkflow
+
+registry = Registry()
+worker_service = WorkerService(registry)
+client = Client(registry, )
+
+registry.add_workflow(BasicWorkflow())

basic/main.py

@@ -0,0 +1,82 @@
+import traceback
+
+from flask import Flask, request
+from iwf.iwf_api.models import (
+    WorkflowStateExecuteRequest,
+    WorkflowStateWaitUntilRequest,
+    WorkflowWorkerRpcRequest,
+)
+from iwf.worker_service import (
+    WorkerService,
+)
+
+from basic.basic_workflow import BasicWorkflow
+from basic.iwf_config import client, worker_service
+
+flask_app = Flask(__name__)
+
+
+# http://localhost:8802/basic/start?workflowId=test-1108&inputNum=4
+@flask_app.route("/basic/start")
+def basic_start():
+    workflow_id = request.args["workflowId"]
+    input_num = request.args["inputNum"]
+
+    run_id = client.start_workflow(BasicWorkflow, workflow_id, 3600, int(input_num))
+    return run_id
+
+# http://localhost:8802/basic/appendString?workflowId=test-1108&str=test
+@flask_app.route("/basic/appendString")
+def basic_append_string():
+    workflow_id = request.args["workflowId"]
+    st = request.args["str"]
+    client.invoke_rpc(workflow_id, BasicWorkflow.append_string, st)
+    return st
+
+# http://localhost:8802/basic/approve?workflowId=test-1108
+@flask_app.route("/basic/approve")
+def basic_approve():
+    workflow_id = request.args["workflowId"]
+    client.invoke_rpc(workflow_id, BasicWorkflow.approve)
+    return "done"
+
+@flask_app.route("/")
+def index():
+    return "iwf workflow home"
+
+
+# below are iWF workflow worker APIs to be called by iWF server
+@flask_app.route(WorkerService.api_path_workflow_state_wait_until, methods=["POST"])
+def handle_wait_until():
+    req = WorkflowStateWaitUntilRequest.from_dict(request.json)
+    resp = worker_service.handle_workflow_state_wait_until(req)
+    return resp.to_dict()
+
+
+@flask_app.route(WorkerService.api_path_workflow_state_execute, methods=["POST"])
+def handle_execute():
+    req = WorkflowStateExecuteRequest.from_dict(request.json)
+    resp = worker_service.handle_workflow_state_execute(req)
+    return resp.to_dict()
+
+
+@flask_app.route(WorkerService.api_path_workflow_worker_rpc, methods=["POST"])
+def handle_rpc():
+    req = WorkflowWorkerRpcRequest.from_dict(request.json)
+    resp = worker_service.handle_workflow_worker_rpc(req)
+    return resp.to_dict()
+
+
+# this handler is extremely useful for debugging iWF
+# the WebUI will be able to show you the error with stacktrace
+@flask_app.errorhandler(Exception)
+def internal_error(exception):
+    return traceback.format_exc(), 500
+
+
+def main():
+    flask_app.run(host="0.0.0.0", port=8802)
+
+
+if __name__ == "__main__":
+    main()