How to combine for_each and if_

Question

I am starting to learn liteflow and wonder if it is possible to somehow combine for_each and if_. I tried the following:

from liteflow.core import *


class Hello(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print("Hello")
        return ExecutionResult.next()


class DoStuff(StepBody):

    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print(f"doing stuff...{context.execution_pointer.context_item}")
        return ExecutionResult.next()


class Goodbye(StepBody):
    def run(self, context: StepExecutionContext) -> ExecutionResult:
        print("Goodbye")
        return ExecutionResult.next()


class MyWorkflow(Workflow):

    def id(self):
        return "MyWorkflow"

    def version(self):
        return 1

    def build(self, builder: WorkflowBuilder):
        builder\
            .start_with(Hello)\
            .for_each(lambda data, context: ["abc", "def", "xyz"])\
                .if_(lambda d, c: not d.startswith("x"))\
                    .do(lambda x: x.start_with(DoStuff))\
            .then(Goodbye)


host = configure_workflow_host()
host.register_workflow(MyWorkflow())
host.start()

wid = host.start_workflow("MyWorkflow", 1, None)

input()
host.stop()

This is not working ('NoneType' object has no attribute 'startswith'), if I replace the condition with something like 0 < 1 DoStuff is exceuted once instead of three times without the if_ and even prints the "wrong" output.

I know that I can eliminate the if_ by using a LC in for_each in this simple example ([x for x in ["abc", "def", "xyz"] if not x.startswith("x")]), but imagine a more complex workflow.

So how do I combine for_each and if_ in a way to get access to the loop variable from for_each in if_ and also pass the correct context to do? Or is this impossible and I have to check the condition in the run method of my steps?

How to combine for_each and if_

0 Answers