DotFlow

dotflow.core.dotflow.DotFlow

Import

You can import the Dotflow class directly from dotflow:

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

Example

class dotflow.core.dotflow.Dotflow

config = Config(
    storage=StorageFile()
)

workflow = DotFlow(config=config)

Parameters:

Name	Type	Description	Default
config	Optional[Config]	Configuration class.	None

Attributes:

Name	Type	Description
workflow_id	UUID
task	List[Task]
start	Manager

Source code in dotflow/core/dotflow.py

class DotFlow:
    """
    Import:
        You can import the **Dotflow** class directly from dotflow:

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

    Example:
        `class` dotflow.core.dotflow.Dotflow

            config = Config(
                storage=StorageFile()
            )

            workflow = DotFlow(config=config)

    Args:
        config (Optional[Config]): Configuration class.

    Attributes:
        workflow_id (UUID):

        task (List[Task]):

        start (Manager):
    """

    def __init__(
            self,
            config: Optional[Config] = None
    ) -> None:
        self.workflow_id = uuid4()
        config = config if config else Config()

        self.task = TaskBuilder(
            config=config,
            workflow_id=self.workflow_id
        )

        self.start = partial(
            Manager,
            tasks=self.task.queue,
            workflow_id=self.workflow_id
        )

    def result_task(self):
        """
        Returns:
            list (List[Task]): Returns a list of Task class.
        """
        return self.task.queue

    def result_context(self):
        """
        Returns:
            list (List[Context]): Returns a list of Context class.
        """
        return [task.current_context for task in self.task.queue]

    def result_storage(self):
        """
        Returns:
            list (List[Any]): Returns a list of assorted objects.
        """
        return [task.current_context.storage for task in self.task.queue]

    def result(self):
        return self.task.result()

workflow_id = uuid4() instance-attribute

task = TaskBuilder(config=config, workflow_id=self.workflow_id) instance-attribute

start = partial(Manager, tasks=self.task.queue, workflow_id=self.workflow_id) instance-attribute

result_task()

Returns:

Name	Type	Description
list	List[Task]	Returns a list of Task class.

Source code in dotflow/core/dotflow.py

def result_task(self):
    """
    Returns:
        list (List[Task]): Returns a list of Task class.
    """
    return self.task.queue

result_context()

Returns:

Name	Type	Description
list	List[Context]	Returns a list of Context class.

Source code in dotflow/core/dotflow.py

def result_context(self):
    """
    Returns:
        list (List[Context]): Returns a list of Context class.
    """
    return [task.current_context for task in self.task.queue]

result_storage()

Returns:

Name	Type	Description
list	List[Any]	Returns a list of assorted objects.

Source code in dotflow/core/dotflow.py

def result_storage(self):
    """
    Returns:
        list (List[Any]): Returns a list of assorted objects.
    """
    return [task.current_context.storage for task in self.task.queue]