From b4ec13818d2c76b1598ec3e3dfb0d60fc3e32d0c Mon Sep 17 00:00:00 2001 From: AdityaByte Date: Tue, 6 Jan 2026 14:46:59 +0530 Subject: [PATCH] feat: Implemented OR based dependency execution, tests and updated docs --- README.rst | 3 +- docs/documentation/task_dep.rst | 124 ++++++++++++++---- pravaha/core/executor.py | 41 +++++- pravaha/core/task.py | 18 ++- pravaha/dependency/__init__.py | 0 pravaha/dependency/dependency.py | 10 ++ pravaha/exception/task.py | 3 + pravaha/utils/dep_resolver.py | 2 +- pravaha/utils/execution_plan.py | 4 +- pravaha/validation/dag.py | 4 +- .../integration/test_dependency_resolution.py | 84 ++++++++++++ tests/unit/core/test_executor.py | 16 +++ tests/unit/core/test_task_init.py | 13 +- 13 files changed, 280 insertions(+), 42 deletions(-) create mode 100644 pravaha/dependency/__init__.py create mode 100644 pravaha/dependency/dependency.py diff --git a/README.rst b/README.rst index 7203e46..74a37e3 100644 --- a/README.rst +++ b/README.rst @@ -9,7 +9,8 @@ Features -------- - Define tasks using the ``@Task`` decorator -- Automatic execution order based on task dependencies +- Automatically resolves task execution order based on defined dependencies +- Supports both OR and AND dependency types - Built-in logging and execution context for task outputs - Task states tracking: ``PENDING``, ``SUCCESS``, ``FAILED``, ``SKIPPED`` - Fail-fast error handling for dependent tasks diff --git a/docs/documentation/task_dep.rst b/docs/documentation/task_dep.rst index fb59419..0258092 100644 --- a/docs/documentation/task_dep.rst +++ b/docs/documentation/task_dep.rst @@ -1,50 +1,126 @@ Task Dependencies -=================== +================= In many workflows, tasks depend on the results of other tasks. -**Task Orchestrator** allows you to define dependencies easily -using the ``depends_on`` attribute. +**Pravaha** allows you to define task dependencies using the ``depends_on`` attribute. -This ensures that tasks are executed in the correct order and -prevents errors caused by missing prerequisites. +Defining dependencies ensures that tasks are executed in the correct order and +prevents failures caused by missing prerequisites. --- Defining Dependencies ------------------------ +--------------------- -You can specify dependencies as a list of task names when -defining a task: +Pravaha supports two types of dependencies: + +1. **OR-based dependencies** +2. **AND-based dependencies** + +--- + +OR-based Dependencies +--------------------- + +An **OR-based dependency** means that a task will execute if **at least one** +of its dependent tasks completes successfully. + +Example +^^^^^^^ .. code-block:: python - from pravaha.core.task import Task - from pravaha.core.executor import TaskExecutor + from pravaha.core.task import Task + from pravaha.core.executor import TaskExecutor + from pravaha.dependency.dependency import Dependency + + # Dependency(type="OR | AND", dependencies=list[str]) + @Task("a", depends_on=Dependency("OR", dependencies=["b", "c"])) + def a(): + print("a") + + @Task("b") + def b(): + pass - @Task(name="task1", depends_on=["task2"]) - def task1(): - print("Task1 executed") + @Task("c") + def c(): + pass - @Task(name="task2") - def task2(): - print("Task2 executed") + TaskExecutor.execute() - if __name__ == "__main__": - TaskExecutor.execute() +--- + +AND-based Dependencies +---------------------- + +An **AND-based dependency** means that a task will execute **only if all** +of its dependent tasks complete successfully. + +AND-based dependencies can be defined in two ways. + +### 1. Implicit AND dependency using a list + +By default, passing a list to ``depends_on`` is treated as an **AND dependency**. + +.. code-block:: python + + from pravaha.core.task import Task + from pravaha.core.executor import TaskExecutor + + @Task(name="task1", depends_on=["task2"]) + def task1(): + print("Task1 executed") + + @Task(name="task2") + def task2(): + print("Task2 executed") + + if __name__ == "__main__": + TaskExecutor.execute() Expected Output ------------------ +^^^^^^^^^^^^^^^ .. code-block:: text - Task2 executed - Task1 executed + Task2 executed + Task1 executed + +--- + +### 2. Explicit AND dependency using ``Dependency`` + +You can also explicitly define an AND-based dependency using the ``Dependency`` object. + +.. code-block:: python + + from pravaha.core.task import Task + from pravaha.core.executor import TaskExecutor + from pravaha.dependency.dependency import Dependency + + # Dependency(type="OR | AND", dependencies=list[str]) + @Task("a", depends_on=Dependency("AND", dependencies=["b", "c"])) + def a(): + print("a") + + @Task("b") + def b(): + pass + + @Task("c") + def c(): + pass + + TaskExecutor.execute() --- Notes -------- +----- - **Task registration order does not matter** when dependencies are defined. -- The executor automatically determines the **correct execution order** based on the dependency graph. -- Circular dependencies will result in an error. Make sure your workflow graph is **acyclic**. +- The executor automatically determines the **correct execution order** + based on the dependency graph. +- Circular dependencies are not allowed. Ensure that your workflow graph + is **acyclic (DAG)**. diff --git a/pravaha/core/executor.py b/pravaha/core/executor.py index 1f332a7..4b290bb 100644 --- a/pravaha/core/executor.py +++ b/pravaha/core/executor.py @@ -3,12 +3,13 @@ from pravaha.enums.task_status import TaskStatus from pravaha.validation.dag import DAGValidator from pravaha.context.condition.context import ConditionContext -from datetime import datetime -from time import time, sleep -import os from pravaha.core.task import ErrorInformation from pravaha.utils.utilities import sort_task_on_the_basis_of_priority, filter_tasks_on_the_basis_of_tags from pravaha.utils.dep_resolver import resolve_dependencies +from pravaha.exception.task import InvalidDependencyType +from datetime import datetime +from time import time, sleep +import os class TaskExecutor: """ @@ -53,11 +54,27 @@ def _execute_helper(cls, task: Task): if task.state in [TaskStatus.SUCCESS, TaskStatus.FAILED, TaskStatus.SKIPPED]: return - for dep_name in task.depends_on: - dep_task = cls.tasks[dep_name] + dep_type = task.depends_on.type + + if dep_type not in ['AND', 'OR', ""]: + raise InvalidDependencyType(f"Invalid dependency type: {dep_type} with task: {task.name}") + + for dep_name in task.depends_on.dependencies: + dep_task = cls.tasks[dep_name] cls._execute_helper(dep_task) + if dep_type == "OR" and dep_task.state == TaskStatus.SUCCESS: + break + + can_we_proceed = True + + if dep_type == "OR": + if not any(cls.tasks[dep].state == TaskStatus.SUCCESS for dep in task.depends_on.dependencies): + can_we_proceed = False + elif dep_type == "AND": + if any(cls.tasks[dep].state in [TaskStatus.FAILED, TaskStatus.SKIPPED] for dep in task.depends_on.dependencies): + can_we_proceed = False - if any(cls.tasks[dep].state in [TaskStatus.FAILED, TaskStatus.SKIPPED] for dep in task.depends_on): + if not can_we_proceed: task.state = TaskStatus.SKIPPED return @@ -68,7 +85,17 @@ def _execute_helper(cls, task: Task): return # Preparing inputs for dependency outputs. - inputs = [cls.ExecutionContext.get(dep_name) for dep_name in task.depends_on] + + inputs = [] + + if dep_type == "OR": + for dep_name in task.depends_on.dependencies: + dep_task = cls.tasks[dep_name] + if dep_task.state == TaskStatus.SUCCESS: + inputs.append(cls.ExecutionContext.get(dep_name)) + + else: + inputs = [cls.ExecutionContext.get(dep_name) for dep_name in task.depends_on.dependencies] # Executing task along with checking for retry. # Implementing retry logic. diff --git a/pravaha/core/task.py b/pravaha/core/task.py index 7b770e1..2516af3 100644 --- a/pravaha/core/task.py +++ b/pravaha/core/task.py @@ -8,6 +8,8 @@ from pravaha.enums.task_status import TaskStatus from pravaha.retry.policy import RetryPolicy from pravaha.enums.task_priority import TaskPriority +from pravaha.dependency.dependency import Dependency +from typing import Union, List, Optional class Task: """ @@ -22,9 +24,8 @@ class Task: tag (str): Tag to the task. """ - def __init__(self, name, depends_on=None, retries: RetryPolicy=None, condition=None, priority=TaskPriority.NORMAL, tag=None): + def __init__(self, name, depends_on: Optional[Union[Dependency, List[str]]] = None, retries: Optional[RetryPolicy]=None, condition=None, priority=TaskPriority.NORMAL, tag=None): self.name = name - self.depends_on = depends_on or [] self.retries = retries self.function_ref = None self.state = TaskStatus.PENDING # Default state @@ -38,6 +39,19 @@ def __init__(self, name, depends_on=None, retries: RetryPolicy=None, condition=N self.condition = condition self.priority = priority self.tag = tag + self.depends_on = self._normalize_dependency(depends_on) + + @staticmethod + def _normalize_dependency(depends_on: Optional[Union[Dependency, List[str]]]) -> Dependency: + + if depends_on is None: + return Dependency() + elif isinstance(depends_on, Dependency): + return depends_on + elif isinstance(depends_on, list): + return Dependency("AND", depends_on) + else: + raise TypeError(f"depends_on must be dependency or List[str]") def __call__(self, original_function): diff --git a/pravaha/dependency/__init__.py b/pravaha/dependency/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/pravaha/dependency/dependency.py b/pravaha/dependency/dependency.py new file mode 100644 index 0000000..96a3e2d --- /dev/null +++ b/pravaha/dependency/dependency.py @@ -0,0 +1,10 @@ +class Dependency: + def __init__(self, type: str = "", dependencies: list[str] = []): + self.type = type + self.dependencies = dependencies + + def get_type(self): + return self.name + + def get_dependencies(self): + return self.dependencies \ No newline at end of file diff --git a/pravaha/exception/task.py b/pravaha/exception/task.py index aa7e5f9..f9fd7dc 100644 --- a/pravaha/exception/task.py +++ b/pravaha/exception/task.py @@ -2,4 +2,7 @@ class TaskFailedError(Exception): pass class TaskNotFoundError(Exception): + pass + +class InvalidDependencyType(Exception): pass \ No newline at end of file diff --git a/pravaha/utils/dep_resolver.py b/pravaha/utils/dep_resolver.py index 332b29d..00135a3 100644 --- a/pravaha/utils/dep_resolver.py +++ b/pravaha/utils/dep_resolver.py @@ -41,7 +41,7 @@ def dfs(task_name: str): visited.add(task_name) task: Task = registry[task_name] - for dep_name in task.depends_on: + for dep_name in task.depends_on.dependencies: dfs(dep_name) resolved[task_name] = task # Adding resolved task to dictionary. diff --git a/pravaha/utils/execution_plan.py b/pravaha/utils/execution_plan.py index bda00c9..011fb27 100644 --- a/pravaha/utils/execution_plan.py +++ b/pravaha/utils/execution_plan.py @@ -16,8 +16,8 @@ def dry_run(): tasks = Registry.get_task() for task in tasks.values(): - if task.depends_on is not None and task.depends_on != []: - final_output.append(f"{task.name} - depends on ({','.join(task.depends_on)})") + if task.depends_on is not None and task.depends_on.dependencies != []: + final_output.append(f"{task.name} - depends on ({','.join(task.depends_on.dependencies)}) with dependency-type: {task.depends_on.type}") else: final_output.append(f"{task.name} - no dependencies ") diff --git a/pravaha/validation/dag.py b/pravaha/validation/dag.py index ab5d4a9..8de74f2 100644 --- a/pravaha/validation/dag.py +++ b/pravaha/validation/dag.py @@ -9,7 +9,7 @@ def validate(tasks: dict): # Missing dependency check. for task in tasks.values(): - for dep in task.depends_on: + for dep in task.depends_on.dependencies: if dep not in tasks: raise MissingDependencyError(f"Task '{task.name}' depends on missing task '{dep}'") @@ -19,7 +19,7 @@ def dfs(task_name, path): path_visited.add(task_name) path.append(task_name) - for dep in tasks[task_name].depends_on: + for dep in tasks[task_name].depends_on.dependencies: if dep not in visited: dfs(dep, path) elif dep in path_visited: diff --git a/tests/integration/test_dependency_resolution.py b/tests/integration/test_dependency_resolution.py index f5a0771..ad4916e 100644 --- a/tests/integration/test_dependency_resolution.py +++ b/tests/integration/test_dependency_resolution.py @@ -3,6 +3,8 @@ from pravaha.enums.task_status import TaskStatus from pravaha.core.registry import Registry from pravaha.exception.task import TaskNotFoundError +from pravaha.exception.validation import MissingDependencyError +from pravaha.dependency.dependency import Dependency import pytest def test_selecting_task_includes_all_dependencies(): @@ -69,3 +71,85 @@ def a(): # The task b is missing so it will raise an error. TaskExecutor.execute(taskgroup=("a", "b")) +def test_task_a_succeeds_if_any_or_dependency_succeeds(): + + @Task("a", depends_on=Dependency("OR", ['b', 'c'])) + def a(val: str): + print("a", val, sep="\n") + + @Task("b") + def b(): + raise ValueError("Value error") + + @Task("c") + def c(): + return "c" + + TaskExecutor.execute() + + tasks = Registry.get_task() + + assert tasks['a'].state == TaskStatus.SUCCESS + assert tasks['b'].state == TaskStatus.FAILED + assert tasks['c'].state == TaskStatus.SUCCESS + + assert TaskExecutor.ExecutionContext.get('c') == "c" + +def test_and_dependency_fails_if_one_fails(): + + @Task("a", depends_on=Dependency("AND", ['b', 'c'])) + def a(val: str): + print("a", val, sep="\n") + + @Task("b") + def b(): + raise ValueError("Value error") + + @Task("c") + def c(): + return "c" + + TaskExecutor.execute() + + tasks = Registry.get_task() + + assert tasks['a'].state == TaskStatus.SKIPPED + assert tasks['b'].state == TaskStatus.FAILED + assert tasks['c'].state == TaskStatus.SUCCESS + +def test_or_dependency_raises_missing_dependecy_error(): + + @Task("a", depends_on=Dependency("OR", ["b"])) + def a(): + pass + + with pytest.raises(MissingDependencyError): + TaskExecutor.execute() + +def test_or_dependency_skips_task_when_all_dependencies_fail(): + + @Task("a", depends_on=Dependency("OR", ['b', 'c'])) + def a(): + pass + + @Task("b") + def b(): + raise TypeError("type-error") + + @Task("c") + def c(): + raise ValueError("value-error") + + TaskExecutor.execute() + + tasks = Registry.get_task() + + assert tasks['a'].state == TaskStatus.SKIPPED + assert tasks['b'].state == TaskStatus.FAILED + assert tasks['c'].state == TaskStatus.FAILED + + assert tasks['b'].error.get_error_msg() == "type-error" + assert tasks['c'].error.get_error_msg() == "value-error" + + assert issubclass(tasks['b'].error.get_error_type(), TypeError) + assert issubclass(tasks['c'].error.get_error_type(), ValueError) \ No newline at end of file diff --git a/tests/unit/core/test_executor.py b/tests/unit/core/test_executor.py index ab9255c..0ad3098 100644 --- a/tests/unit/core/test_executor.py +++ b/tests/unit/core/test_executor.py @@ -6,6 +6,9 @@ from pravaha.retry.policy import RetryPolicy from pravaha.retry.backoff import fixed_delay from pravaha.enums.task_priority import TaskPriority +from pravaha.dependency.dependency import Dependency +from pravaha.exception.task import InvalidDependencyType +import pytest def test_single_task_execution_success(): @@ -180,3 +183,16 @@ def task3(): TaskExecutor.execute() assert execution_order == ['task1', 'task3', 'task2'] + +def test_executor_raises_error_on_invalid_dependency_type(): + + @Task("a", depends_on=Dependency("INVALID", ['b'])) + def a(): + pass + + @Task("b") + def b(): + pass + + with pytest.raises(InvalidDependencyType): + TaskExecutor.execute() \ No newline at end of file diff --git a/tests/unit/core/test_task_init.py b/tests/unit/core/test_task_init.py index 2858a16..d3c7e8c 100644 --- a/tests/unit/core/test_task_init.py +++ b/tests/unit/core/test_task_init.py @@ -3,6 +3,8 @@ """ from pravaha.core.task import Task from pravaha.enums.task_status import TaskStatus +from pravaha.dependency.dependency import Dependency +import pytest def test_task_initializes_with_required_name(): task = Task(name="demo_task") @@ -10,7 +12,7 @@ def test_task_initializes_with_required_name(): def test_task_defaults_when_optional_args_not_provided(): task = Task(name="demo_task") - assert task.depends_on == [] + assert isinstance(task.depends_on, Dependency) assert task.state == TaskStatus.PENDING assert task.error is None assert task.start_time == "" @@ -22,10 +24,15 @@ def test_task_defaults_when_optional_args_not_provided(): def test_task_accepts_dependency_list(): task = Task("demo_task", depends_on=['task1', 'task2']) dependency_set = set(['task1', 'task2']) - assert dependency_set == set(task.depends_on) + assert dependency_set == set(task.depends_on.dependencies) def test_task_does_not_share_depends_on_between_instances(): task1 = Task('task1', depends_on=['task3']) task2 = Task('task2', depends_on=['task3']) - assert task1.depends_on is not task2.depends_on \ No newline at end of file + assert task1.depends_on is not task2.depends_on + +def test_task_raises_type_error_on_invalid_depends_on_argument(): + + with pytest.raises(TypeError): + task1 = Task("task1", depends_on=tuple()) \ No newline at end of file