Manager

dotflow.core.workflow.Manager

Import

You can import the Manager class with:

from dotflow.core.workflow import Manager

Example

class dotflow.core.workflow.Manager

workflow = Manager(
    tasks=[tasks],
    on_success=basic_callback,
    on_failure=basic_callback,
    keep_going=True
)

Parameters:

Name	Type	Description	Default
tasks	List[Task]	A list containing objects of type Task.	required
on_success	Callable	Success function to be executed after the completion of the entire workflow. It's essentially a callback for successful scenarios.	basic_callback
on_failure	Callable	Failure function to be executed after the completion of the entire workflow. It's essentially a callback for error scenarios	basic_callback
mode	TypeExecution	Parameter that defines the execution mode of the workflow. Currently, there are options to execute in sequential, background, or parallel mode. The sequential mode is used by default.	SEQUENTIAL
keep_going	bool	A parameter that receives a boolean object with the purpose of continuing or not the execution of the workflow in case of an error during the execution of a task. If it is true, the execution will continue; if it is False, the workflow will stop.	False
workflow_id	UUID	Workflow ID.	None
resume	bool	If True, checks for existing checkpoints before executing each task. Tasks with a checkpoint are skipped and their saved context is used. Requires a persistent storage provider and a fixed workflow_id.	False

Attributes:

Name	Type	Description
on_success	Callable
on_failure	Callable
workflow_id	UUID
started	datetime

Source code in dotflow/core/workflow.py

class Manager:
    """
    Import:
        You can import the **Manager** class with:

            from dotflow.core.workflow import Manager

    Example:
        `class` dotflow.core.workflow.Manager

            workflow = Manager(
                tasks=[tasks],
                on_success=basic_callback,
                on_failure=basic_callback,
                keep_going=True
            )

    Args:
        tasks (List[Task]):
            A list containing objects of type Task.

        on_success (Callable):
            Success function to be executed after the completion of the entire
            workflow. It's essentially a callback for successful scenarios.

        on_failure (Callable):
            Failure function to be executed after the completion of the entire
            workflow. It's essentially a callback for error scenarios

        mode (TypeExecution):
            Parameter that defines the execution mode of the workflow. Currently,
            there are options to execute in **sequential**, **background**, or **parallel** mode.
            The sequential mode is used by default.


        keep_going (bool):
            A parameter that receives a boolean object with the purpose of continuing
            or not the execution of the workflow in case of an error during the
            execution of a task. If it is **true**, the execution will continue;
            if it is **False**, the workflow will stop.

        workflow_id (UUID): Workflow ID.

        resume (bool):
            If True, checks for existing checkpoints before executing each task.
            Tasks with a checkpoint are skipped and their saved context is used.
            Requires a persistent storage provider and a fixed workflow_id.

    Attributes:
        on_success (Callable):

        on_failure (Callable):

        workflow_id (UUID):

        started (datetime):
    """

    def __init__(
        self,
        tasks: list[Task],
        on_success: Callable = basic_callback,
        on_failure: Callable = basic_callback,
        mode: TypeExecution = TypeExecution.SEQUENTIAL,
        keep_going: bool = False,
        workflow_id: UUID = None,
        resume: bool = False,
        on_input_change: str = "reuse",
        fingerprint: str | None = None,
        config=None,
    ) -> None:
        self.tasks = tasks
        self.on_success = on_success
        self.on_failure = on_failure
        self.workflow_id = workflow_id or uuid4()
        self.started = datetime.now()
        self.config = config

        if on_input_change not in VALID_POLICIES:
            raise InvalidOnInputChange(value=on_input_change)

        self.on_input_change = on_input_change
        self.fingerprint = fingerprint

        if resume and self.config:
            self._enforce_input_fingerprint(tasks=tasks)

        if self.config:
            self.config.tracer.start_workflow(
                workflow_id=self.workflow_id, mode=mode, tasks_count=len(tasks)
            )
            self.config.metrics.workflow_started(
                workflow_id=self.workflow_id, mode=mode
            )
            self.config.server.update_workflow(
                workflow=self.workflow_id,
                status=WorkflowStatus.IN_PROGRESS,
            )

        execution = None
        groups = grouper(tasks=tasks)

        VALID_MODES = {
            "sequential",
            "sequential_group",
            "background",
            "parallel",
        }
        if mode not in VALID_MODES:
            raise ExecutionModeNotExist()

        execution = getattr(self, mode)

        self.tasks = execution(
            tasks=tasks,
            workflow_id=self.workflow_id,
            ignore=keep_going,
            groups=groups,
            resume=resume,
        )

        if mode != TypeExecution.BACKGROUND:
            self._callback_workflow(tasks=self.tasks)
        elif self.config:

            def _background_cleanup():
                self.thread.join()
                self._callback_workflow(tasks=self.tasks)

            threading.Thread(target=_background_cleanup, daemon=True).start()

    def _enforce_input_fingerprint(self, tasks: list[Task]) -> None:
        storage = self.config.storage
        workflow_key = str(self.workflow_id)

        if self.fingerprint is not None:
            current_fp = self.fingerprint
        else:
            initial_payloads = [
                getattr(task.initial_context, "storage", None)
                for task in tasks
            ]
            current_fp = fingerprint_of(initial_payloads)

        stored_fp = read_fingerprint(storage=storage, workflow_id=workflow_key)

        if stored_fp is None:
            write_fingerprint(
                storage=storage,
                workflow_id=workflow_key,
                value=current_fp,
            )
            return

        if stored_fp == current_fp:
            return

        if self.on_input_change == "reuse":
            return

        if self.on_input_change == "raise":
            raise InputChangedError(workflow_id=workflow_key)

        storage.clear(workflow_id=workflow_key)
        write_fingerprint(
            storage=storage,
            workflow_id=workflow_key,
            value=current_fp,
        )

    def _callback_workflow(self, tasks: list[Task]):
        duration = (datetime.now() - self.started).total_seconds()
        final_status = [task.status for task in tasks]
        failed = TypeStatus.FAILED in final_status

        if self.config:
            self.config.tracer.end_workflow(
                workflow_id=self.workflow_id,
                duration=duration,
                failed=failed,
            )
            if failed:
                self.config.metrics.workflow_failed(self.workflow_id, duration)
            else:
                self.config.metrics.workflow_completed(
                    self.workflow_id, duration
                )

        if self.config:
            wf_status = (
                WorkflowStatus.FAILED if failed else WorkflowStatus.COMPLETED
            )
            self.config.server.update_workflow(
                workflow=self.workflow_id,
                status=wf_status,
            )

        if failed:
            self.on_failure(tasks=tasks)
        else:
            self.on_success(tasks=tasks)

    def sequential(self, **kwargs) -> list[Task]:
        """Run tasks sequentially. Auto-selects SequentialGroup when multiple groups exist."""
        if len(kwargs.get("groups", {})) > 1:
            process = SequentialGroup(**kwargs)
            return process.get_tasks()

        process = Sequential(**kwargs)
        return process.get_tasks()

    def sequential_group(self, **kwargs) -> list[Task]:
        """Run task groups sequentially, with tasks within each group also sequential."""
        process = SequentialGroup(**kwargs)
        return process.get_tasks()

    def background(self, **kwargs) -> list[Task]:
        """Run tasks in a background thread."""
        process = Background(**kwargs)
        self.thread = process.thread
        return process.get_tasks()

    def parallel(self, **kwargs) -> list[Task]:
        """Run task groups in parallel using multiprocessing."""
        process = Parallel(**kwargs)
        return process.get_tasks()

_callback_workflow(tasks)

Source code in dotflow/core/workflow.py

def _callback_workflow(self, tasks: list[Task]):
    duration = (datetime.now() - self.started).total_seconds()
    final_status = [task.status for task in tasks]
    failed = TypeStatus.FAILED in final_status

    if self.config:
        self.config.tracer.end_workflow(
            workflow_id=self.workflow_id,
            duration=duration,
            failed=failed,
        )
        if failed:
            self.config.metrics.workflow_failed(self.workflow_id, duration)
        else:
            self.config.metrics.workflow_completed(
                self.workflow_id, duration
            )

    if self.config:
        wf_status = (
            WorkflowStatus.FAILED if failed else WorkflowStatus.COMPLETED
        )
        self.config.server.update_workflow(
            workflow=self.workflow_id,
            status=wf_status,
        )

    if failed:
        self.on_failure(tasks=tasks)
    else:
        self.on_success(tasks=tasks)

sequential(**kwargs)

Run tasks sequentially. Auto-selects SequentialGroup when multiple groups exist.

Source code in dotflow/core/workflow.py

def sequential(self, **kwargs) -> list[Task]:
    """Run tasks sequentially. Auto-selects SequentialGroup when multiple groups exist."""
    if len(kwargs.get("groups", {})) > 1:
        process = SequentialGroup(**kwargs)
        return process.get_tasks()

    process = Sequential(**kwargs)
    return process.get_tasks()

sequential_group(**kwargs)

Run task groups sequentially, with tasks within each group also sequential.

Source code in dotflow/core/workflow.py

def sequential_group(self, **kwargs) -> list[Task]:
    """Run task groups sequentially, with tasks within each group also sequential."""
    process = SequentialGroup(**kwargs)
    return process.get_tasks()

background(**kwargs)

Run tasks in a background thread.

Source code in dotflow/core/workflow.py

def background(self, **kwargs) -> list[Task]:
    """Run tasks in a background thread."""
    process = Background(**kwargs)
    self.thread = process.thread
    return process.get_tasks()

parallel(**kwargs)

Run task groups in parallel using multiprocessing.

Source code in dotflow/core/workflow.py

def parallel(self, **kwargs) -> list[Task]:
    """Run task groups in parallel using multiprocessing."""
    process = Parallel(**kwargs)
    return process.get_tasks()