Skip to content

Output: result_storage

workflow.result_storage() returns only the payloads (Context.storage) of each task — no metadata. Use it when downstream code consumes raw data and metadata is noise.

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 storage in workflow.result_storage():
        print(storage)
        assert storage == "ok"

    return workflow


if __name__ == "__main__":
    main()

Class step

A class-based step yields a list of nested Context objects in storage. Iterate twice to reach the inner payloads.

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 storages in workflow.result_storage():
        for storage in storages:
            print(storage.time, storage.storage)
            assert storage.storage

    return workflow


if __name__ == "__main__":
    main()

When to use

  • Pure data hand-off to a downstream system (DB insert, API push)
  • Asserting payload shape in tests without unwrapping Context
  • Forwarding step output to a non-dotflow consumer