⚙️ FEATURE: Data type improvement

Author: FernandoCelmer

dotflow/abc/logs.py

@@ -1,10 +1,11 @@
 """Logs ABC"""
 
 from typing import Literal
-from enum import StrEnum
 
 from abc import ABC, abstractmethod
 
+from dotflow.core.types.enum import StrEnum
+
 
 class TypeLog(StrEnum):
 

dotflow/core/types/enum.py

@@ -0,0 +1,18 @@
+"""Enum Type module"""
+
+import sys
+
+from enum import Enum
+
+
+class StrEnumType(str, Enum):
+    ...
+
+
+if sys.version_info >= (3, 11):
+    from enum import StrEnum as StrEnum
+else:
+    StrEnum = StrEnumType
+
+
+__all__ = ["StrEnum"]

dotflow/core/types/execution.py

@@ -1,8 +1,9 @@
 """Execution Mode Type module"""
 
-from enum import StrEnum
 from typing import Literal
 
+from dotflow.core.types.enum import StrEnum
+
 
 class ExecutionModeType(StrEnum):
     """

dotflow/core/types/status.py

@@ -1,8 +1,9 @@
 """Status Task Type module"""
 
-from enum import StrEnum
 from typing_extensions import Literal
 
+from dotflow.core.types.enum import StrEnum
+
 
 class StatusTaskType(StrEnum):
     """
@@ -19,18 +20,6 @@ class StatusTaskType(StrEnum):
     RETRY = "Retry"
     FAILED = "Failed"
 
-    @classmethod
-    def get_symbol(cls, value: str) -> str:
-        status = {
-           StatusTaskType.NOT_STARTED: "⚪",
-           StatusTaskType.IN_PROGRESS: "🔵",
-           StatusTaskType.COMPLETED: "✅",
-           StatusTaskType.PAUSED: "◼️",
-           StatusTaskType.RETRY: "❗",
-           StatusTaskType.FAILED: "❌"
-        }
-        return status.get(value)
-
 
 TYPE_STATUS_TASK = Literal[
     StatusTaskType.NOT_STARTED,

dotflow/core/types/storage.py

@@ -1,8 +1,9 @@
 """Storage Type module"""
 
-from enum import StrEnum
 from typing import Literal
 
+from dotflow.core.types.enum import StrEnum
+
 
 class StorageType(StrEnum):
     """

dotflow/core/types/worflow.py

@@ -1,8 +1,9 @@
 """Workflow Status Type module"""
 
-from enum import StrEnum
 from typing import Literal
 
+from dotflow.core.types.enum import StrEnum
+
 
 class WorkflowStatusType(StrEnum):
     """
@@ -21,4 +22,4 @@ class WorkflowStatusType(StrEnum):
     WorkflowStatusType.NEW,
     WorkflowStatusType.IN_PROGRESS,
     WorkflowStatusType.COMPLETED
-]
+]

dotflow/providers/notify_telegram.py

@@ -10,6 +10,18 @@
 from dotflow.logging import logger
 
 
+def get_symbol(value: str) -> str:
+    status = {
+        StatusTaskType.NOT_STARTED: "⚪",
+        StatusTaskType.IN_PROGRESS: "🔵",
+        StatusTaskType.COMPLETED: "✅",
+        StatusTaskType.PAUSED: "◼️",
+        StatusTaskType.RETRY: "❗",
+        StatusTaskType.FAILED: "❌"
+    }
+    return status.get(value)
+
+
 class NotifyTelegram(Notify):
 
     MESSAGE = "{symbol} {status}\n```json\n{task}```\n{workflow_id}-{task_id}"
@@ -50,7 +62,7 @@ def send(self, task: Any) -> None:
 
     def _get_text(self, task: Any) -> str:
         return self.MESSAGE.format(
-            symbol=StatusTaskType.get_symbol(task.status),
+            symbol=get_symbol(task.status),
             status=task.status,
             workflow_id=task.workflow_id,
             task_id=task.task_id,

dotflow/types.py

@@ -1,9 +1,14 @@
 """Types module"""
 
+from dotflow.core.types.execution import ExecutionModeType
 from dotflow.core.types.status import StatusTaskType
 from dotflow.core.types.storage import StorageType
+from dotflow.core.types.worflow import WorkflowStatusType
+
 
 __all__ = [
+    "ExecutionModeType",
     "StatusTaskType",
-    "StorageType"
+    "StorageType",
+    "WorkflowStatusType"
 ]