diff --git a/docs/nav/concepts/concept-task-engine.md b/docs/nav/concepts/concept-task-engine.md new file mode 100644 index 00000000..8a37e458 --- /dev/null +++ b/docs/nav/concepts/concept-task-engine.md @@ -0,0 +1,107 @@ +# Task Engine + +The `TaskEngine` manages the **lifecycle** of a single task — status transitions, duration tracking, retry, timeout, backoff, error handling, and tracer integration. Execution strategies (`Sequential`, `Parallel`, etc.) are responsible only for **ordering and parallelism**. + +## Architecture + +```mermaid +sequenceDiagram + participant S as Strategy + participant E as TaskEngine + participant A as @action + + S->>E: TaskEngine(task, workflow_id, previous_context) + S->>E: with engine.start() + E->>E: status = IN_PROGRESS + E->>E: tracer.start_task() + S->>E: engine.execute_with_retry() + + loop retry loop (max_attempts) + alt timeout > 0 + E->>E: _execute_with_timeout() + else no timeout + E->>E: _execute_single() + end + E->>A: task.step() + alt success + A-->>E: Context + E-->>S: result + else failure & attempts remaining + E->>E: status = RETRY + E->>E: sleep(delay) + end + end + + E->>E: status = COMPLETED / FAILED + E->>E: tracer.end_task() + S->>S: task.callback() + S->>S: _flow_callback() +``` + +## How it works + +The engine uses a **context manager** pattern to separate lifecycle from execution: + +```python +engine = TaskEngine(task=task, workflow_id=workflow_id, previous_context=previous_context) + +with engine.start(): + engine.execute_with_retry() +``` + +### `start()` — lifecycle context manager + +Manages everything that happens **around** the execution: + +```mermaid +stateDiagram-v2 + [*] --> IN_PROGRESS: start() + IN_PROGRESS --> COMPLETED: success + IN_PROGRESS --> RETRY: retry attempt + RETRY --> COMPLETED: success after retry + RETRY --> FAILED: max attempts reached + IN_PROGRESS --> FAILED: exception + COMPLETED --> [*]: end_task tracer + FAILED --> [*]: end_task tracer +``` + +1. Sets `status = IN_PROGRESS` and starts the timer +2. Starts the tracer span +3. **Yields** — the execution block runs here +4. On success: sets `duration` and `status = COMPLETED` +5. On error: sets `errors` and `status = FAILED` +6. Always: ends the tracer span + +### `execute_with_retry()` — retry, timeout, and backoff + +Reads `retry`, `timeout`, `retry_delay`, and `backoff` from the `@action` decorator and manages the full retry loop: + +```mermaid +flowchart TD + A["execute_with_retry()"] --> B{"timeout > 0?"} + B -->|yes| C["_execute_with_timeout()"] + B -->|no| D["_execute_single()"] + C --> E{"success?"} + D --> E + E -->|yes| F["return result"] + E -->|no| G{"attempt < max?"} + G -->|yes| H["status = RETRY\nsleep(delay)"] + H -->|"backoff?"| I["delay *= 2"] + I --> B + H -->|no backoff| B + G -->|no| J["raise exception"] +``` + +- If `timeout > 0`: uses `ThreadPoolExecutor` with a real deadline +- If execution fails and `attempt < max_attempts`: sets `status = RETRY`, waits, and retries +- If `backoff = True`: doubles the delay after each failed attempt + +### `execute()` — single execution + +Calls the task function once without retry. Used internally by `execute_with_retry()` and available for cases where retry is not needed. + +## References + +- [Task lifecycle and status](concept-task-lifecycle.md) +- [`@action` decorator](../reference/action.md) +- [`TypeStatus`](../reference/type-status.md) diff --git a/docs/nav/concepts/concept-task-lifecycle.md b/docs/nav/concepts/concept-task-lifecycle.md index 50eaac9a..affed896 100644 --- a/docs/nav/concepts/concept-task-lifecycle.md +++ b/docs/nav/concepts/concept-task-lifecycle.md @@ -11,7 +11,7 @@ Every task moves through **status** values tracked by Dotflow. They describe whe ## Retries and pauses -- **`RETRY`** — A retry is scheduled (for example after backoff); thread-safe retry behavior is part of the action runner. +- **`RETRY`** — A retry is scheduled (for example after backoff); retry behavior is managed by the `TaskEngine`. - **`PAUSED`** — Execution is held (depending on workflow configuration and provider behavior). Retry policy is configured per task (timeouts, backoff, etc.); see [Task retry](../tutorial/task-retry.md) and [Task backoff](../tutorial/task-backoff.md). diff --git a/docs/nav/tutorial/tracer-opentelemetry.md b/docs/nav/tutorial/tracer-opentelemetry.md index 2db18efa..b92d3043 100644 --- a/docs/nav/tutorial/tracer-opentelemetry.md +++ b/docs/nav/tutorial/tracer-opentelemetry.md @@ -61,8 +61,8 @@ Run the example and open [http://localhost:16686](http://localhost:16686) to see | ABC method | When | Span action | |------------|------|-------------| | `start_workflow` | Manager.__init__ | Creates parent span | -| `start_task` | Execution.__init__ | Creates child span | -| `end_task` | Execution finally | Sets attributes, status, ends span | +| `start_task` | TaskEngine.start() | Creates child span | +| `end_task` | TaskEngine.start() finally | Sets attributes, status, ends span | | `end_workflow` | _callback_workflow | Sets workflow status, ends parent span | ## Compatible backends diff --git a/docs_src/checkpoint/checkpoint.py b/docs_src/checkpoint/checkpoint.py index ca7ee880..f878a1bc 100644 --- a/docs_src/checkpoint/checkpoint.py +++ b/docs_src/checkpoint/checkpoint.py @@ -1,3 +1,5 @@ +from uuid import UUID + from dotflow import Config, DotFlow, action from dotflow.providers import StorageFile @@ -21,7 +23,7 @@ def step_three(previous_context): def main(): - workflow = DotFlow(config=config, workflow_id="my-etl-pipeline") + workflow = DotFlow(config=config, workflow_id=UUID("12345678-1234-5678-1234-567812345678")) workflow.task.add(step=step_one) workflow.task.add(step=step_two) diff --git a/dotflow/abc/flow.py b/dotflow/abc/flow.py index 95f22d04..f2b6d4f4 100644 --- a/dotflow/abc/flow.py +++ b/dotflow/abc/flow.py @@ -3,7 +3,9 @@ from abc import ABC, abstractmethod from uuid import UUID +from dotflow.core.context import Context from dotflow.core.task import Task +from dotflow.core.types import TypeStatus class Flow(ABC): @@ -40,3 +42,24 @@ def _flow_callback(self, task: Task) -> None: @abstractmethod def run(self) -> None: return None + + def _has_checkpoint(self, task: Task) -> bool: + if not self.resume: + return False + + context = task.config.storage.get( + key=task.config.storage.key(task=task) + ) + + return context.storage is not None + + def _restore_checkpoint(self, task: Task) -> Context: + previous_context = task.config.storage.get( + key=task.config.storage.key(task=task) + ) + + task.status = TypeStatus.COMPLETED + task.current_context = previous_context + self._flow_callback(task=task) + + return previous_context diff --git a/dotflow/core/action.py b/dotflow/core/action.py index cec7824f..fa26682b 100644 --- a/dotflow/core/action.py +++ b/dotflow/core/action.py @@ -2,13 +2,10 @@ import asyncio from collections.abc import Callable -from concurrent.futures import ThreadPoolExecutor -from time import sleep from types import FunctionType from dotflow.core.context import Context -from dotflow.core.exception import ExecutionWithClassError, TaskError -from dotflow.core.types.status import TypeStatus +from dotflow.core.exception import ExecutionWithClassError def is_execution_with_class_internal_error(error: Exception) -> bool: @@ -103,13 +100,13 @@ def __call__(self, *args, **kwargs): if contexts: return Context( - storage=self._run_action(*args, task=task, **contexts), + storage=self._run_action(*args, **contexts), task_id=task.task_id, workflow_id=task.workflow_id, ) return Context( - storage=self._run_action(*args, task=task), + storage=self._run_action(*args), task_id=task.task_id, workflow_id=task.workflow_id, ) @@ -123,77 +120,33 @@ def action(*_args, **_kwargs): if contexts: return Context( - storage=self._run_action(*_args, task=task, **contexts), + storage=self._run_action(*_args, **contexts), task_id=task.task_id, workflow_id=task.workflow_id, ) return Context( - storage=self._run_action(*_args, task=task), + storage=self._run_action(*_args), task_id=task.task_id, workflow_id=task.workflow_id, ) - return action + action.retry = self.retry + action.timeout = self.timeout + action.retry_delay = self.retry_delay + action.backoff = self.backoff - def _run_action(self, *args, task=None, **kwargs): - current_delay = self.retry_delay + return action + def _run_action(self, *args, **kwargs): is_async = asyncio.iscoroutinefunction(self.func) - max_attempts = max(1, self.retry) - - for attempt in range(1, max_attempts + 1): - try: - if self.timeout: - executor = ThreadPoolExecutor(max_workers=1) - try: - future = executor.submit( - self._call_func, - is_async, - *args, - **kwargs, - ) - result = future.result(timeout=self.timeout) - except TimeoutError: - future.cancel() - executor.shutdown(wait=False, cancel_futures=True) - raise - except Exception: - executor.shutdown(wait=False) - raise - else: - executor.shutdown(wait=False) - else: - result = self._call_func(is_async, *args, **kwargs) - - return result - - except TimeoutError: - raise - - except Exception as error: - last_exception = error - - if is_execution_with_class_internal_error( - error=last_exception - ): - raise ExecutionWithClassError() from None - - if attempt == max_attempts: - raise last_exception from None - - if task is not None: - task.retry_count += 1 - task.errors = TaskError( - error=error, - attempt=attempt, - ) - task.status = TypeStatus.RETRY - - sleep(current_delay) - if self.backoff: - current_delay *= 2 + try: + return self._call_func(is_async, *args, **kwargs) + except Exception as error: + if is_execution_with_class_internal_error(error=error): + raise ExecutionWithClassError() from None + raise def _call_func(self, is_async, *args, **kwargs): if is_async: diff --git a/dotflow/core/engine.py b/dotflow/core/engine.py new file mode 100644 index 00000000..1e94400a --- /dev/null +++ b/dotflow/core/engine.py @@ -0,0 +1,277 @@ +"""TaskEngine module""" + +import re +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from contextlib import contextmanager +from datetime import datetime +from inspect import getsourcelines +from time import sleep +from types import FunctionType +from uuid import UUID + +try: + from types import NoneType +except ImportError: + NoneType = type(None) + +from dotflow.core.action import Action +from dotflow.core.context import Context +from dotflow.core.exception import ExecutionWithClassError, TaskError +from dotflow.core.task import Task +from dotflow.core.types import TypeStatus +from dotflow.logging import logger + + +class TaskEngine: + """Manages the execution lifecycle of a single task. + + Separates the task lifecycle (status, duration, error handling) + from how tasks are executed (sequential, parallel, background). + """ + + VALID_OBJECTS = [ + str, + int, + float, + complex, + dict, + list, + tuple, + set, + frozenset, + range, + bool, + FunctionType, + NoneType, + bytes, + bytearray, + memoryview, + ] + + def __init__( + self, + task: Task, + workflow_id: UUID, + previous_context: Context = None, + ) -> None: + self.task = task + self.workflow_id = workflow_id + self.previous_context = previous_context + self._start_time = None + + @contextmanager + def start(self): + """Prepares the task for execution and manages its lifecycle.""" + self.task.workflow_id = self.workflow_id + self.task.previous_context = self.previous_context + self.task.config.tracer.start_task(task=self.task) + self.task.status = TypeStatus.IN_PROGRESS + self._start_time = datetime.now() + + try: + yield self + except AssertionError as err: + raise err + except Exception as err: + self.task.errors = err + self.task.current_context = None + self.task.status = TypeStatus.FAILED + else: + self.task.duration = ( + datetime.now() - self._start_time + ).total_seconds() + if self.task.status in ( + TypeStatus.IN_PROGRESS, + TypeStatus.RETRY, + ): + self.task.status = TypeStatus.COMPLETED + finally: + self.task.config.tracer.end_task(task=self.task) + + def execute(self): + """Executes the task function and returns the context.""" + current_context = self.task.step( + initial_context=self.task.initial_context, + previous_context=self.task.previous_context, + task=self.task, + ) + + if type(current_context.storage) not in self.VALID_OBJECTS: + current_context = self._execution_with_class( + class_instance=current_context.storage + ) + + self.task.current_context = current_context + return current_context + + def execute_with_retry(self): + """Executes the task with retry, timeout, and backoff managed by the engine. + + Reads retry, timeout, retry_delay, and backoff from the task's step + (the @action decorator) and manages the full retry loop. + """ + step = self.task.step + max_attempts = max(1, step.retry) + timeout = step.timeout + retry_delay = step.retry_delay + backoff = step.backoff + current_delay = retry_delay + + for attempt in range(1, max_attempts + 1): + try: + if timeout: + result = self._execute_with_timeout(timeout) + else: + result = self._execute_single() + + self.task.current_context = result + return result + + except TimeoutError: + raise + + except Exception as error: + if self._is_class_internal_error(error): + raise + + if attempt == max_attempts: + raise + + self.task.retry_count += 1 + self.task.errors = TaskError( + error=error, + attempt=attempt, + ) + self.task.status = TypeStatus.RETRY + + sleep(current_delay) + if backoff: + current_delay *= 2 + + def _execute_single(self): + """Executes the task function once, handling class-based steps.""" + current_context = self.task.step( + initial_context=self.task.initial_context, + previous_context=self.task.previous_context, + task=self.task, + ) + + if type(current_context.storage) not in self.VALID_OBJECTS: + current_context = self._execution_with_class( + class_instance=current_context.storage + ) + + return current_context + + def _execute_with_timeout(self, seconds: int): + """Executes the task function with a real timeout using ThreadPoolExecutor.""" + executor = ThreadPoolExecutor(max_workers=1) + try: + future = executor.submit(self._execute_single) + return future.result(timeout=seconds) + except TimeoutError: + future.cancel() + executor.shutdown(wait=False, cancel_futures=True) + raise + except Exception: + executor.shutdown(wait=False) + raise + + @staticmethod + def _is_class_internal_error(error: Exception) -> bool: + """Checks if an error is an internal class execution error.""" + message = str(error) + patterns = [ + "initial_context", + "previous_context", + "missing 1 required positional argument: 'self'", + ] + return any(pattern in message for pattern in patterns) + + def _is_action(self, class_instance: Callable, func: Callable): + try: + return ( + callable(getattr(class_instance, func)) + and getattr(class_instance, func).__module__ + == Action.__module__ + and not func.startswith("__") + ) + except AttributeError: + return False + + def _execution_orderer( + self, callable_list: list[str], class_instance: Callable + ) -> tuple[int, Callable]: + ordered_list = [] + + try: + inside_code = getsourcelines(class_instance.__class__)[0] + + for callable_name in callable_list: + pattern = re.compile( + rf"\bdef\s+{re.escape(callable_name)}\s*\(" + ) + for index, code in enumerate(inside_code): + if pattern.search(code): + ordered_list.append((index, callable_name)) + break + + ordered_list.sort() + return ordered_list + + except TypeError as err: + logger.error( + "Internal problem with ordering the class functions, " + "but don't worry, it was executed.: %s", + str(err), + ) + + for index, callable_name in enumerate(callable_list): + ordered_list.append((index, callable_name)) + + return ordered_list + + def _execution_with_class(self, class_instance: Callable): + new_context = Context(storage=[]) + previous_context = self.task.previous_context + callable_list = [ + func + for func in dir(class_instance) + if self._is_action(class_instance, func) + ] + + ordered_list = self._execution_orderer( + callable_list=callable_list, class_instance=class_instance + ) + + for index, new in enumerate(ordered_list): + new_object = getattr(class_instance, new[1]) + try: + subcontext = new_object( + initial_context=self.task.initial_context, + previous_context=previous_context, + task=self.task, + ) + subcontext.task_id = index + new_context.storage.append(subcontext) + previous_context = subcontext + + except Exception as error: + if not isinstance(error, ExecutionWithClassError): + raise error + + subcontext = new_object( + class_instance, + initial_context=self.task.initial_context, + previous_context=previous_context, + task=self.task, + ) + subcontext.task_id = index + new_context.storage.append(subcontext) + previous_context = subcontext + + if not new_context.storage: + return Context(storage=class_instance) + + return new_context diff --git a/dotflow/core/execution.py b/dotflow/core/execution.py index 8f8e9a7e..565e536b 100644 --- a/dotflow/core/execution.py +++ b/dotflow/core/execution.py @@ -1,45 +1,22 @@ -"""Execution module""" +"""Execution module — compatibility wrapper around TaskEngine""" -import re from collections.abc import Callable -from datetime import datetime -from inspect import getsourcelines -from types import FunctionType from uuid import UUID -try: - from types import NoneType -except ImportError: - NoneType = type(None) - -from dotflow.core.action import Action from dotflow.core.context import Context -from dotflow.core.exception import ExecutionWithClassError +from dotflow.core.engine import TaskEngine from dotflow.core.task import Task -from dotflow.core.types import TypeStatus -from dotflow.logging import logger from dotflow.utils import basic_callback class Execution: - VALID_OBJECTS = [ - str, - int, - float, - complex, - dict, - list, - tuple, - set, - frozenset, - range, - bool, - FunctionType, - NoneType, - bytes, - bytearray, - memoryview, - ] + """Backward-compatible wrapper that delegates to TaskEngine. + + Preserves the original interface where instantiation triggers execution. + New code should use TaskEngine directly. + """ + + VALID_OBJECTS = TaskEngine.VALID_OBJECTS def __init__( self, @@ -49,128 +26,24 @@ def __init__( _flow_callback: Callable = basic_callback, ) -> None: self.task = task - self.task.previous_context = previous_context - self.task.workflow_id = workflow_id - self.task.config.tracer.start_task(task=self.task) - self.task.status = TypeStatus.IN_PROGRESS + engine = TaskEngine( + task=task, + workflow_id=workflow_id, + previous_context=previous_context, + ) - self._execution(_flow_callback) + with engine.start(): + engine.execute_with_retry() + + self.task.callback(task=self.task) + _flow_callback(task=self.task) def _is_action(self, class_instance: Callable, func: Callable): - try: - return ( - callable(getattr(class_instance, func)) - and getattr(class_instance, func).__module__ - == Action.__module__ - and not func.startswith("__") - ) - except AttributeError: - return False + engine = TaskEngine.__new__(TaskEngine) + return engine._is_action(class_instance, func) def _execution_orderer( self, callable_list: list[str], class_instance: Callable - ) -> tuple[int, Callable]: - ordered_list = [] - - try: - inside_code = getsourcelines(class_instance.__class__)[0] - - for callable_name in callable_list: - pattern = re.compile( - rf"\bdef\s+{re.escape(callable_name)}\s*\(" - ) - for index, code in enumerate(inside_code): - if pattern.search(code): - ordered_list.append((index, callable_name)) - break - - ordered_list.sort() - return ordered_list - - except TypeError as err: - logger.error( - "Internal problem with ordering the class functions, but don't worry, it was executed.: %s", - str(err), - ) - - for index, callable_name in enumerate(callable_list): - ordered_list.append((index, callable_name)) - - return ordered_list - - def _execution_with_class(self, class_instance: Callable): - new_context = Context(storage=[]) - previous_context = self.task.previous_context - callable_list = [ - func - for func in dir(class_instance) - if self._is_action(class_instance, func) - ] - - ordered_list = self._execution_orderer( - callable_list=callable_list, class_instance=class_instance - ) - - for index, new in enumerate(ordered_list): - new_object = getattr(class_instance, new[1]) - try: - subcontext = new_object( - initial_context=self.task.initial_context, - previous_context=previous_context, - task=self.task, - ) - subcontext.task_id = index - new_context.storage.append(subcontext) - previous_context = subcontext - - except Exception as error: - if not isinstance(error, ExecutionWithClassError): - raise error - - subcontext = new_object( - class_instance, - initial_context=self.task.initial_context, - previous_context=previous_context, - task=self.task, - ) - subcontext.task_id = index - new_context.storage.append(subcontext) - previous_context = subcontext - - if not new_context.storage: - return Context(storage=class_instance) - - return new_context - - def _execution(self, _flow_callback): - try: - start = datetime.now() - current_context = self.task.step( - initial_context=self.task.initial_context, - previous_context=self.task.previous_context, - task=self.task, - ) - - if type(current_context.storage) not in self.VALID_OBJECTS: - current_context = self._execution_with_class( - class_instance=current_context.storage - ) - - self.task.current_context = current_context - self.task.duration = (datetime.now() - start).total_seconds() - self.task.status = TypeStatus.COMPLETED - - except AssertionError as err: - raise err - - except Exception as err: - self.task.errors = err - self.task.current_context = None - self.task.status = TypeStatus.FAILED - - finally: - self.task.config.tracer.end_task(task=self.task) - self.task.callback(task=self.task) - _flow_callback(task=self.task) - - return self.task + ): + engine = TaskEngine.__new__(TaskEngine) + return engine._execution_orderer(callable_list, class_instance) diff --git a/dotflow/core/workflow.py b/dotflow/core/workflow.py index 805e7026..81b03a1b 100644 --- a/dotflow/core/workflow.py +++ b/dotflow/core/workflow.py @@ -10,6 +10,7 @@ from dotflow.abc.flow import Flow from dotflow.core.context import Context +from dotflow.core.engine import TaskEngine from dotflow.core.exception import ExecutionModeNotExist from dotflow.core.execution import Execution from dotflow.core.task import Task, TaskError @@ -214,37 +215,26 @@ def get_tasks(self) -> list[Task]: def _flow_callback(self, task: Task) -> None: self.queue.append(task) - def _has_checkpoint(self, task: Task) -> bool: - if not self.resume: - return False - - context = task.config.storage.get( - key=task.config.storage.key(task=task) - ) - - return context.storage is not None - def run(self) -> None: previous_context = Context(workflow_id=self.workflow_id) for task in self.tasks: if self._has_checkpoint(task): - previous_context = task.config.storage.get( - key=task.config.storage.key(task=task) - ) - - task.status = TypeStatus.COMPLETED - task.current_context = previous_context - self._flow_callback(task=task) + previous_context = self._restore_checkpoint(task) continue - Execution( + engine = TaskEngine( task=task, workflow_id=self.workflow_id, previous_context=previous_context, - _flow_callback=self._flow_callback, ) + with engine.start(): + engine.execute_with_retry() + + task.callback(task=task) + self._flow_callback(task=task) + previous_context = task.config.storage.get( key=task.config.storage.key(task=task) ) @@ -309,37 +299,26 @@ def run(self) -> None: for process in self._processes: process.join() - def _has_checkpoint(self, task: Task) -> bool: - if not self.resume: - return False - - context = task.config.storage.get( - key=task.config.storage.key(task=task) - ) - - return context.storage is not None - def _run_group(self, groups: list[Task]) -> None: previous_context = Context(workflow_id=self.workflow_id) for task in groups: if self._has_checkpoint(task): - previous_context = task.config.storage.get( - key=task.config.storage.key(task=task) - ) - - task.status = TypeStatus.COMPLETED - task.current_context = previous_context - self._flow_callback(task=task) + previous_context = self._restore_checkpoint(task) continue - Execution( + engine = TaskEngine( task=task, workflow_id=self.workflow_id, previous_context=previous_context, - _flow_callback=self._flow_callback, ) + with engine.start(): + engine.execute_with_retry() + + task.callback(task=task) + self._flow_callback(task=task) + previous_context = task.config.storage.get( key=task.config.storage.key(task=task) ) @@ -362,37 +341,26 @@ def _flow_callback(self, task: Task) -> None: with self._lock: self.queue.append(task) - def _has_checkpoint(self, task: Task) -> bool: - if not self.resume: - return False - - context = task.config.storage.get( - key=task.config.storage.key(task=task) - ) - - return context.storage is not None - def _run_sequential(self) -> None: previous_context = Context(workflow_id=self.workflow_id) for task in self.tasks: if self._has_checkpoint(task): - previous_context = task.config.storage.get( - key=task.config.storage.key(task=task) - ) - - task.status = TypeStatus.COMPLETED - task.current_context = previous_context - self._flow_callback(task=task) + previous_context = self._restore_checkpoint(task) continue - Execution( + engine = TaskEngine( task=task, workflow_id=self.workflow_id, previous_context=previous_context, - _flow_callback=self._flow_callback, ) + with engine.start(): + engine.execute_with_retry() + + task.callback(task=task) + self._flow_callback(task=task) + previous_context = task.config.storage.get( key=task.config.storage.key(task=task) ) diff --git a/mkdocs.yml b/mkdocs.yml index c1c0166a..2fe0cabb 100644 --- a/mkdocs.yml +++ b/mkdocs.yml @@ -214,6 +214,7 @@ nav: - "Concept of context": nav/concepts/concept-of-context.md - Providers: nav/concepts/concept-providers.md - "Task lifecycle and status": nav/concepts/concept-task-lifecycle.md + - "Task engine": nav/concepts/concept-task-engine.md - "Cron overlap": nav/concepts/concept-cron-overlap.md - Process Mode: - nav/concepts/process-mode-sequential.md diff --git a/tests/core/test_action.py b/tests/core/test_action.py index 717ce84c..0044b361 100644 --- a/tests/core/test_action.py +++ b/tests/core/test_action.py @@ -8,7 +8,6 @@ from dotflow.core.action import Action from dotflow.core.context import Context from dotflow.core.task import Task -from dotflow.core.types.status import TypeStatus from tests.mocks import ( action_step, simple_step, @@ -45,22 +44,15 @@ def test_instantiating_action_class_with_retry(self): self.assertEqual(inside.func, simple_step) self.assertIsInstance(decorated_function, Context) - def test_instantiating_action_class_with_fail_retry(self): + def test_instantiating_action_class_with_fail(self): error_message = "Fail!" - number_of_retries = 5 - - inside = Action(simple_step_with_fail, retry=number_of_retries) - with self._caplog.at_level(logging.ERROR): - try: - inside() - except Exception as error: - self.assertEqual(error.args[0], error_message) + inside = Action(simple_step_with_fail) - self.assertEqual(len(self._caplog.records), number_of_retries) + with self.assertRaises(Exception) as ctx: + inside() - for record in self._caplog.records: - self.assertEqual(record.message, error_message) + self.assertEqual(str(ctx.exception), error_message) def test_retry_zero_still_executes_task_once(self): call_count = {"n": 0} @@ -85,22 +77,13 @@ def always_fail(): with self.assertRaises(ValueError): inside(task=self.task) - def test_sets_retry_status_before_retrying(self): - calls = {"count": 0} - statuses = [] - - def flaky_step(): - calls["count"] += 1 - if calls["count"] == 1: - raise Exception("Fail once") - statuses.append(self.task.status) - return "ok" - - inside = Action(flaky_step, retry=2, retry_delay=0) - inside(task=self.task) + def test_action_single_attempt_raises_on_failure(self): + def always_fail(): + raise ValueError("fail") - self.assertEqual(len(statuses), 1) - self.assertEqual(statuses[0], TypeStatus.RETRY) + inside = Action(always_fail) + with self.assertRaises(ValueError): + inside(task=self.task) def test_retry_exception_does_not_chain_to_itself(self): def always_fail(): @@ -117,17 +100,12 @@ def always_fail(): "Exception must not be its own __cause__ (circular chain)", ) - def test_backoff_does_not_mutate_retry_delay(self): - def always_fail(): - raise RuntimeError("fail") - - inside = Action(always_fail, retry=3, retry_delay=1, backoff=True) - - with unittest.mock.patch("dotflow.core.action.sleep"): # noqa: SIM117 - with self.assertRaises(RuntimeError): - inside() + def test_action_preserves_retry_params(self): + inside = Action(simple_step, retry=3, retry_delay=2, backoff=True) - self.assertEqual(inside.retry_delay, 1) + self.assertEqual(inside.retry, 3) + self.assertEqual(inside.retry_delay, 2) + self.assertTrue(inside.backoff) def test_action_class_with_previous_context(self): inside = Action(simple_step_with_previous_context, task=self.task) diff --git a/tests/core/test_engine.py b/tests/core/test_engine.py new file mode 100644 index 00000000..64c659d6 --- /dev/null +++ b/tests/core/test_engine.py @@ -0,0 +1,421 @@ +"""Test TaskEngine""" + +import logging +import unittest +from uuid import uuid4 + +from pytest import fixture # type: ignore + +from dotflow.core.context import Context +from dotflow.core.engine import TaskEngine +from dotflow.core.task import Task +from dotflow.core.types import TypeStatus +from tests.mocks import ( + ActionStep, + ActionStepExecutionOrderer, + ActionStepWithContexts, + ActionStepWithError, + ActionStepWithInitialContext, + ActionStepWithPreviousContext, + SimpleStep, + action_step, + action_step_valid_object, + action_step_with_contexts, + action_step_with_error, + action_step_with_initial_context, + action_step_with_previous_context, + action_step_with_retry, + action_step_with_timeout, + simple_callback, + simple_step, +) + + +class TestTaskEngine(unittest.TestCase): + @fixture(autouse=True) + def inject_fixtures(self, caplog): + self._caplog = caplog + + def setUp(self): + self.workflow_id = uuid4() + self.context = {"context": True} + self.task = Task(task_id=0, step=action_step, callback=simple_callback) + + def test_engine_with_function_completed(self): + workflow_id = uuid4() + task = Task(task_id=0, step=action_step, callback=simple_callback) + engine = TaskEngine( + task=task, workflow_id=workflow_id, previous_context=Context() + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.workflow_id, workflow_id) + self.assertIsNotNone(task.duration) + + def test_engine_with_function_failed(self): + workflow_id = uuid4() + task = Task( + task_id=0, step=action_step_with_error, callback=simple_callback + ) + engine = TaskEngine( + task=task, workflow_id=workflow_id, previous_context=Context() + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.FAILED) + self.assertEqual(task.workflow_id, workflow_id) + + def test_engine_with_class_completed(self): + execution_log = "" + workflow_id = uuid4() + task = Task(task_id=0, step=ActionStep, callback=simple_callback) + engine = TaskEngine( + task=task, workflow_id=workflow_id, previous_context=Context() + ) + + with self._caplog.at_level(logging.NOTSET): + with engine.start(): + engine.execute() + + for log in self._caplog.records: + if log.funcName == "run": + execution_log = log.message + + self.assertEqual(execution_log, "ActionStep: Run function executed") + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.workflow_id, workflow_id) + + def test_engine_with_class_failed(self): + execution_log = "" + workflow_id = uuid4() + task = Task( + task_id=0, step=ActionStepWithError, callback=simple_callback + ) + engine = TaskEngine( + task=task, workflow_id=workflow_id, previous_context=Context() + ) + + with self._caplog.at_level(logging.NOTSET): + with engine.start(): + engine.execute() + + for log in self._caplog.records: + if log.funcName == "run": + execution_log = log.message + + self.assertEqual( + execution_log, "ActionStepWithError: Run function executed" + ) + self.assertEqual(task.status, TypeStatus.FAILED) + self.assertEqual(task.workflow_id, workflow_id) + + def test_engine_function_with_initial_context(self): + task = Task( + task_id=0, + step=action_step_with_initial_context, + callback=simple_callback, + initial_context=self.context, + ) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=None + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.initial_context.storage, self.context) + + def test_engine_function_with_previous_context(self): + task = Task( + task_id=0, + step=action_step_with_previous_context, + callback=simple_callback, + ) + engine = TaskEngine( + task=task, + workflow_id=self.workflow_id, + previous_context=self.context, + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task._previous_context.storage, self.context) + + def test_engine_function_with_contexts(self): + task = Task( + task_id=0, + step=action_step_with_contexts, + callback=simple_callback, + initial_context=self.context, + ) + engine = TaskEngine( + task=task, + workflow_id=self.workflow_id, + previous_context=self.context, + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.initial_context.storage, self.context) + self.assertEqual(task.previous_context.storage, self.context) + + def test_engine_class_with_initial_context(self): + task = Task( + task_id=0, + step=ActionStepWithInitialContext, + callback=simple_callback, + initial_context=self.context, + ) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=None + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.initial_context.storage, self.context) + + def test_engine_class_with_previous_context(self): + task = Task( + task_id=0, + step=ActionStepWithPreviousContext, + callback=simple_callback, + ) + engine = TaskEngine( + task=task, + workflow_id=self.workflow_id, + previous_context=self.context, + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.previous_context.storage, self.context) + self.assertEqual( + task.current_context.storage[0].storage, {"func": "run_x"} + ) + self.assertEqual( + task.current_context.storage[1].storage, {"func": "run_y"} + ) + + def test_engine_class_with_contexts(self): + task = Task( + task_id=0, + step=ActionStepWithContexts, + callback=simple_callback, + initial_context=self.context, + ) + engine = TaskEngine( + task=task, + workflow_id=self.workflow_id, + previous_context=self.context, + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.initial_context.storage, self.context) + self.assertEqual(task.previous_context.storage, self.context) + self.assertEqual( + task.current_context.storage[0].storage, {"foo": "bar"} + ) + + def test_engine_is_action_true(self): + engine = TaskEngine( + task=self.task, + workflow_id=self.workflow_id, + previous_context=Context(), + ) + + with engine.start(): + engine.execute() + + class_instance = ActionStep(task=engine.task).storage + self.assertTrue( + engine._is_action(class_instance=class_instance, func="run") + ) + + def test_engine_is_action_false(self): + engine = TaskEngine( + task=self.task, + workflow_id=self.workflow_id, + previous_context=Context(), + ) + + with engine.start(): + engine.execute() + + class_instance = SimpleStep() + self.assertFalse( + engine._is_action(class_instance=class_instance, func="run") + ) + + def test_engine_is_action_init_false(self): + engine = TaskEngine( + task=self.task, + workflow_id=self.workflow_id, + previous_context=Context(), + ) + + with engine.start(): + engine.execute() + + class_instance = SimpleStep() + self.assertFalse( + engine._is_action(class_instance=class_instance, func="__init__") + ) + + def test_engine_execution_orderer(self): + expected_value = [ + (3, "func_f"), + (7, "func_e"), + (11, "func_d"), + (15, "func_c"), + (19, "func_b"), + (23, "func_a"), + ] + + engine = TaskEngine( + task=self.task, + workflow_id=self.workflow_id, + previous_context=Context(), + ) + + with engine.start(): + engine.execute() + + class_instance = ActionStepExecutionOrderer(task=engine.task).storage + callable_list = [ + func + for func in dir(class_instance) + if engine._is_action(class_instance, func) + ] + + self.assertListEqual( + engine._execution_orderer( + callable_list=callable_list, class_instance=class_instance + ), + expected_value, + ) + + def test_engine_valid_objects(self): + valid_objects = [ + "", + 1, + 1.0, + complex(3, 5), + {}, + [], + (1, 2, 3), + {1, 2, 3}, + frozenset({1, 2, 3}), + range(5), + simple_step, + True, + None, + b"Hello", + bytearray(5), + memoryview(bytes(5)), + ] + + for input_value in valid_objects: + task = Task( + task_id=0, + step=action_step_valid_object, + callback=simple_callback, + initial_context=input_value, + ) + + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=None + ) + + with engine.start(): + engine.execute() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + self.assertEqual(task.current_context.storage, input_value) + + def test_engine_status_in_progress_during_execution(self): + captured_status = None + + task = Task(task_id=0, step=action_step, callback=simple_callback) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=Context() + ) + + with engine.start(): + captured_status = task.status + engine.execute() + + self.assertEqual(captured_status, TypeStatus.IN_PROGRESS) + self.assertEqual(task.status, TypeStatus.COMPLETED) + + def test_engine_context_manager_without_execute(self): + task = Task(task_id=0, step=action_step, callback=simple_callback) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=Context() + ) + + with engine.start(): + pass + + self.assertEqual(task.status, TypeStatus.COMPLETED) + + def test_engine_execute_with_retry_completed(self): + task = Task( + task_id=0, + step=action_step_with_retry, + callback=simple_callback, + ) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=Context() + ) + + with engine.start(): + engine.execute_with_retry() + + self.assertEqual(task.status, TypeStatus.COMPLETED) + + def test_engine_execute_with_retry_failed(self): + task = Task( + task_id=0, + step=action_step_with_error, + callback=simple_callback, + ) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=Context() + ) + + with engine.start(): + engine.execute_with_retry() + + self.assertEqual(task.status, TypeStatus.FAILED) + + def test_engine_execute_with_timeout(self): + task = Task( + task_id=0, + step=action_step_with_timeout, + callback=simple_callback, + ) + engine = TaskEngine( + task=task, workflow_id=self.workflow_id, previous_context=Context() + ) + + with engine.start(): + engine.execute_with_retry() + + self.assertEqual(task.status, TypeStatus.COMPLETED) diff --git a/tests/mocks/__init__.py b/tests/mocks/__init__.py index 3cb764ef..2bec3eb3 100644 --- a/tests/mocks/__init__.py +++ b/tests/mocks/__init__.py @@ -15,11 +15,13 @@ from tests.mocks.step_function import ( action_step, action_step_valid_object, + action_step_with_backoff, action_step_with_contexts, action_step_with_error, action_step_with_initial_context, action_step_with_previous_context, action_step_with_retry, + action_step_with_timeout, simple_step, simple_step_with_fail, simple_step_with_initial_context, @@ -43,6 +45,8 @@ "action_step_with_previous_context", "action_step_with_contexts", "action_step_with_retry", + "action_step_with_backoff", + "action_step_with_timeout", "action_step_with_error", "simple_step", "simple_step_with_params", diff --git a/tests/mocks/step_function.py b/tests/mocks/step_function.py index f0abefdf..15acc685 100644 --- a/tests/mocks/step_function.py +++ b/tests/mocks/step_function.py @@ -41,6 +41,16 @@ def action_step_with_retry() -> Any: return {"foo": "bar"} +@action(retry=3, retry_delay=0, backoff=True) +def action_step_with_backoff() -> Any: + raise RuntimeError("always fail") + + +@action(timeout=10) +def action_step_with_timeout() -> Any: + return {"timed": True} + + @action def action_step_with_error() -> Any: raise Exception("Fail!")