Contexto Anterior

Enviando Contexto

Para enviar um contexto, basta retornar alguma informação na implementação da sua tarefa. Você pode retornar qualquer objeto, e ele fará parte do contexto.

Python 3.8+

from dotflow import DotFlow, action, Context


@action
def extract_task():
    print("extract")
    return "extract"


@action
def transform_task(previous_context: Context):
    print(previous_context.storage, "transform")
    assert previous_context.storage == "extract"

    return "transform"


@action
def load_task(previous_context: Context):
    print(previous_context.storage, "load")
    assert previous_context.storage == "transform"

    return "load"


def main():
    workflow = DotFlow()

    workflow.task.add(step=extract_task)
    workflow.task.add(step=transform_task)
    workflow.task.add(step=load_task)

    workflow.start()

    return workflow


if __name__ == "__main__":
    main()

Recebendo Contexto

Neste exemplo, um valor de contexto anterior é recebido a partir do atributo previous_contexto definido na função.

Python 3.8+

from dotflow import DotFlow, action, Context


@action
def extract_task():
    print("extract")
    return "extract"


@action
def transform_task(previous_context: Context):
    print(previous_context.storage, "transform")
    assert previous_context.storage == "extract"

    return "transform"


@action
def load_task(previous_context: Context):
    print(previous_context.storage, "load")
    assert previous_context.storage == "transform"

    return "load"


def main():
    workflow = DotFlow()

    workflow.task.add(step=extract_task)
    workflow.task.add(step=transform_task)
    workflow.task.add(step=load_task)

    workflow.start()

    return workflow


if __name__ == "__main__":
    main()