Ir para o conteúdo

Output: result_context

workflow.result_context() returns the list of Context objects produced by every task in the queue. Use it when you need both the payload (storage) and the metadata (time, task_id, workflow_id) of each step.

Function step

from dotflow import DotFlow, action


@action
def simple_step():
    return "ok"


def main():
    workflow = DotFlow()

    workflow.task.add(step=simple_step)
    workflow.start()

    for context in workflow.result_context():
        print(context.time, context.storage)
        assert context.storage == "ok"

    return workflow


if __name__ == "__main__":
    main()

Class step

A class-based step returns one Context whose storage is a list of nested contexts — one per inner @action method.

from dotflow import DotFlow, action


@action(retry=5)
class Step:
    def auxiliary_function(self):
        """This function will not be executed, because
        it does not have an 'action' decorator.
        """

    @action
    def first_function(self):
        return {"foo": "bar"}

    @action
    def second_function(self):
        return True


def main():
    workflow = DotFlow()

    workflow.task.add(step=Step)
    workflow.start()

    for contexts in workflow.result_context():
        for context in contexts.storage:
            print(context.time, context.storage)
            assert context.storage

    return workflow


if __name__ == "__main__":
    main()

When to use

  • Auditing — full timeline of what each step produced and when
  • Tracing — task_id and workflow_id propagate through the context chain
  • Debugging — inspect intermediate state without rerunning