Merge pull request #90 from aravindhbalaji04/feature/set-retry-status

⚙️ FEATURE: Set task status to RETRY during retry attempts

Author: FernandoCelmer

.code_quality/ruff.toml

@@ -43,6 +43,7 @@ ignore = [
     "B027",   # Empty method in ABC without abstractmethod (intentional in base providers)
     "ARG001", # Unused function argument (intentional in interface implementations)
     "ARG002", # Unused method argument (intentional in *args/**kwargs constructors and providers)
+    "UP007",  # Use X | Y for unions — disabled due to Python 3.9 lacking support for this syntax
 ]
 
 # Allowed configurations for specific rules

LAST_VERSION

@@ -1 +1 @@
-0.13.2.dev2
+0.13.2

dotflow/__init__.py

@@ -1,6 +1,6 @@
 """Dotflow __init__ module."""
 
-__version__ = "0.13.2.dev2"
+__version__ = "0.13.2"
 __description__ = "🎲 Dotflow turns an idea into flow!"
 
 from .core.action import Action as action

dotflow/cli/validators/start.py

@@ -2,14 +2,16 @@
 
 from __future__ import annotations
 
+from typing import Optional
+
 from pydantic import BaseModel, Field  # type: ignore
 
 from dotflow.settings import Settings as settings
 
 
 class StartValidator(BaseModel):
     step: str
-    callable: str | None = Field(default=None)
-    initial_context: str | None = Field(default=None)
-    output: bool | None = Field(default=True)
-    path: str | None = Field(default=settings.START_PATH)
+    callable: Optional[str] = Field(default=None)
+    initial_context: Optional[str] = Field(default=None)
+    output: Optional[bool] = Field(default=True)
+    path: Optional[str] = Field(default=settings.START_PATH)

dotflow/core/action.py

@@ -7,6 +7,7 @@
 
 from dotflow.core.context import Context
 from dotflow.core.exception import ExecutionWithClassError
+from dotflow.core.types.status import TypeStatus
 
 
 def is_execution_with_class_internal_error(error: Exception) -> bool:
@@ -97,54 +98,62 @@ def __call__(self, *args, **kwargs):
         if self.func:
             self._set_params()
 
-            task = self._get_task(kwargs=kwargs)
+            self._current_task = self._get_task(kwargs=kwargs)
             contexts = self._get_context(kwargs=kwargs)
 
             if contexts:
                 return Context(
                     storage=self._run_action(*args, **contexts),
-                    task_id=task.task_id,
-                    workflow_id=task.workflow_id,
+                    task_id=self._current_task.task_id,
+                    workflow_id=self._current_task.workflow_id,
                 )
 
             return Context(
                 storage=self._run_action(*args),
-                task_id=task.task_id,
-                workflow_id=task.workflow_id,
+                task_id=self._current_task.task_id,
+                workflow_id=self._current_task.workflow_id,
             )
 
         # No parameters
         def action(*_args, **_kwargs):
             self.func = args[0]
             self._set_params()
 
-            task = self._get_task(kwargs=_kwargs)
+            self._current_task = self._get_task(kwargs=_kwargs)
             contexts = self._get_context(kwargs=_kwargs)
 
             if contexts:
                 return Context(
                     storage=self._run_action(*_args, **contexts),
-                    task_id=task.task_id,
-                    workflow_id=task.workflow_id,
+                    task_id=self._current_task.task_id,
+                    workflow_id=self._current_task.workflow_id,
                 )
 
             return Context(
                 storage=self._run_action(*_args),
-                task_id=task.task_id,
-                workflow_id=task.workflow_id,
+                task_id=self._current_task.task_id,
+                workflow_id=self._current_task.workflow_id,
             )
 
         return action
 
     def _run_action(self, *args, **kwargs):
+        task = getattr(self, '_current_task', None)
+
         for attempt in range(1, self.retry + 1):
             try:
                 if self.timeout:
                     with ThreadPoolExecutor(max_workers=1) as executor:
                         future = executor.submit(self.func, *args, **kwargs)
-                        return future.result(timeout=self.timeout)
+                        result = future.result(timeout=self.timeout)
+                else:
+                    result = self.func(*args, **kwargs)
+
+                # Reset status to IN_PROGRESS after successful retry
+                if task and attempt > 1:
+                    task.status = TypeStatus.IN_PROGRESS
 
-                return self.func(*args, **kwargs)
+                return result
 
             except Exception as error:
                 last_exception = error
@@ -157,6 +166,9 @@ def _run_action(self, *args, **kwargs):
                 if attempt == self.retry:
                     raise last_exception from last_exception
 
+                if task:
+                    task.status = TypeStatus.RETRY
+
                 sleep(self.retry_delay)
                 if self.backoff:
                     self.retry_delay *= 2

dotflow/core/serializers/task.py

@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import json
-from typing import Any
+from typing import Any, Optional
 from uuid import UUID
 
 from pydantic import BaseModel, ConfigDict, Field, field_validator
@@ -20,16 +20,16 @@ class SerializerTask(BaseModel):
     model_config = ConfigDict(title="task")
 
     task_id: int = Field(default=None)
-    workflow_id: UUID | None = Field(default=None)
+    workflow_id: Optional[UUID] = Field(default=None)
     status: str = Field(default=None, alias="_status")
-    error: SerializerTaskError | None = Field(default=None, alias="_error")
-    duration: float | None = Field(default=None, alias="_duration")
+    error: Optional[SerializerTaskError] = Field(default=None, alias="_error")
+    duration: Optional[float] = Field(default=None, alias="_duration")
     initial_context: Any = Field(default=None, alias="_initial_context")
     current_context: Any = Field(default=None, alias="_current_context")
     previous_context: Any = Field(default=None, alias="_previous_context")
     group_name: str = Field(default=None)
-    max: int | None = Field(default=None, exclude=True)
-    size_message: str | None = Field(
+    max: Optional[int] = Field(default=None, exclude=True)
+    size_message: Optional[str] = Field(
         default="Context size exceeded", exclude=True
     )
 

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "dotflow"
-version = "0.13.2.dev2"
+version = "0.13.2"
 authors = [
   { name="Fernando Celmer", email="email@fernandocelmer.com" },
 ]
@@ -37,7 +37,7 @@ mongodb = ["dotflow-mongodb"]
 
 [tool.poetry]
 name = "dotflow"
-version = "0.13.2.dev2"
+version = "0.13.2"
 description = "🎲 Dotflow turns an idea into flow!"
 authors = ["Fernando Celmer <email@fernandocelmer.com>"]
 readme = "README.md"

tests/core/decorators/test_time.py

@@ -27,7 +27,9 @@ def run():
         self.assertGreaterEqual(task.duration, 0)
 
     def test_return_value_is_preserved(self):
-        expected_task = Task(task_id=0, step=action_step, callback=simple_callback)
+        expected_task = Task(
+            task_id=0, step=action_step, callback=simple_callback
+        )
 
         @time
         def run():

tests/core/test_action.py

@@ -8,6 +8,8 @@
 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,
@@ -61,6 +63,27 @@ def test_instantiating_action_class_with_fail_retry(self):
             for record in self._caplog.records:
                 self.assertEqual(record.message, error_message)
 
+    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")
+            # On second attempt, capture status (should be RETRY)
+            statuses.append(self.task.status)
+            return "ok"
+
+        inside = Action(flaky_step, retry=2, retry_delay=0)
+        inside(task=self.task)
+
+        # Verify status was RETRY during the retry attempt
+        self.assertEqual(len(statuses), 1)
+        self.assertEqual(statuses[0], TypeStatus.RETRY)
+        # Verify status was reset to IN_PROGRESS after successful retry
+        self.assertEqual(self.task.status, TypeStatus.IN_PROGRESS)
+
     def test_action_class_with_previous_context(self):
         inside = Action(simple_step_with_previous_context, task=self.task)
 

tests/providers/test_api_default.py

@@ -74,7 +74,9 @@ def test_none_returns_none(self):
         self.assertIsNone(ApiDefault._callable_path(None))
 
     def test_string_returns_string(self):
-        self.assertEqual(ApiDefault._callable_path("my.module.func"), "my.module.func")
+        self.assertEqual(
+            ApiDefault._callable_path("my.module.func"), "my.module.func"
+        )
 
     def test_callable_returns_module_name(self):
         def my_func():
@@ -146,7 +148,9 @@ def test_calls_post_when_ready(self):
         mock_response = MagicMock()
         mock_response.raise_for_status = MagicMock()
 
-        with patch("dotflow.providers.api_default.post", return_value=mock_response):
+        with patch(
+            "dotflow.providers.api_default.post", return_value=mock_response
+        ):
             api.create_workflow("workflow-id")
 
         mock_response.raise_for_status.assert_called_once()
@@ -158,7 +162,9 @@ def test_logs_error_on_exception(self):
             user_token="token",
         )
 
-        with patch("dotflow.providers.api_default.post", side_effect=Exception("fail")):
+        with patch(
+            "dotflow.providers.api_default.post", side_effect=Exception("fail")
+        ):
             result = api.create_workflow("workflow-id")
 
         self.assertIsNone(result)