Skip to content

Storage Provider

The storage provider defines where task outputs are persisted.

Available providers

  • StorageDefault: keeps data in memory during workflow execution.
  • StorageFile: persists output to disk (good for local runs and debugging).

Example

from dotflow import Config, DotFlow, action
from dotflow.providers import StorageDefault, StorageFile


@action
def task():
    return {"hello": "dotflow"}


def main():
    # In-memory storage (default-like behavior).
    workflow_memory = DotFlow(config=Config(storage=StorageDefault()))
    workflow_memory.task.add(step=task)
    workflow_memory.start()

    # File storage persisted in ".output/tasks".
    workflow_file = DotFlow(config=Config(storage=StorageFile(path=".output")))
    workflow_file.task.add(step=task)
    workflow_file.start()

    return workflow_memory, workflow_file


if __name__ == "__main__":
    main()
👀 Full file preview 
from dotflow import Config, DotFlow, action
from dotflow.providers import StorageDefault, StorageFile


@action
def task():
    return {"hello": "dotflow"}


def main():
    # In-memory storage (default-like behavior).
    workflow_memory = DotFlow(config=Config(storage=StorageDefault()))
    workflow_memory.task.add(step=task)
    workflow_memory.start()

    # File storage persisted in ".output/tasks".
    workflow_file = DotFlow(config=Config(storage=StorageFile(path=".output")))
    workflow_file.task.add(step=task)
    workflow_file.start()

    return workflow_memory, workflow_file


if __name__ == "__main__":
    main()

Extra examples

File storage only:

from dotflow import Config, DotFlow, action
from dotflow.providers import StorageFile


@action
def simple_step():
    return "ok"


def main():
    config = Config(storage=StorageFile(path=".output"))

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

    return workflow


if __name__ == "__main__":
    main()
👀 Full file preview 
from dotflow import Config, DotFlow, action
from dotflow.providers import StorageFile


@action
def simple_step():
    return "ok"


def main():
    config = Config(storage=StorageFile(path=".output"))

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

    return workflow


if __name__ == "__main__":
    main()

References