🎨 format: move to use from_trace method.

Author: korawica

src/ddeutil/workflow/conf.py

@@ -117,7 +117,7 @@ def trace_handlers(self) -> list[dict[str, Any]]:
     def debug(self) -> bool:
         """Debug flag for echo log that use DEBUG mode.
 
-        :rtype: bool
+        Returns: bool
         """
         return str2bool(env("LOG_DEBUG_MODE", "true"))
 
@@ -472,9 +472,11 @@ def dynamic(
 def pass_env(value: T) -> T:  # pragma: no cov
     """Passing environment variable to an input value.
 
-    :param value: (Any) A value that want to pass env var searching.
+    Args:
+        value (Any): A value that want to pass env var searching.
 
-    :rtype: Any
+    Returns:
+        Any: An any value that have passed environment variable.
     """
     if isinstance(value, dict):
         return {k: pass_env(value[k]) for k in value}

src/ddeutil/workflow/result.py

@@ -283,13 +283,19 @@ class Info(TypedDict):
     exec_latency: NotRequired[float]
 
 
+class System(TypedDict):
+    __sys_release_dryrun_mode: NotRequired[bool]
+    __sys_exec_break_circle: NotRequired[str]
+
+
 class Context(TypedDict):
     """Context dict typed."""
 
     status: Status
+    info: Info
+    sys: NotRequired[System]
     context: NotRequired[DictData]
     errors: NotRequired[Union[list[ErrorData], ErrorData]]
-    info: NotRequired[DictData]
 
 
 class Layer(str, Enum):

src/ddeutil/workflow/stages.py

@@ -398,14 +398,11 @@ def execute(
                 f"💥 Error Failed:||🚨 {traceback.format_exc()}||",
                 module="stage",
             )
-            return Result(
-                run_id=run_id,
-                parent_run_id=parent_run_id,
+            return Result.from_trace(trace).catch(
                 status=FAILED,
                 context=catch(
                     context, status=FAILED, updated={"errors": to_dict(e)}
                 ),
-                extras=self.extras,
             )
         finally:
             context["info"].update(
@@ -857,25 +854,20 @@ async def axecute(
                     f"[STAGE]: 🤫 Stage Failed with disable traceback:||{e}"
                 )
             st: Status = get_status_from_error(e)
-            return Result(
-                run_id=run_id,
-                parent_run_id=parent_run_id,
+            return Result.from_trace(trace).catch(
                 status=st,
                 context=catch(context, status=st, updated=updated),
-                extras=self.extras,
             )
         except Exception as e:
             await trace.aerror(
-                f"[STAGE]:💥 Error Failed:||🚨 {traceback.format_exc()}||"
+                f"💥 Error Failed:||🚨 {traceback.format_exc()}||",
+                module="stage",
             )
-            return Result(
-                run_id=run_id,
-                parent_run_id=parent_run_id,
+            return Result.from_trace(trace).catch(
                 status=FAILED,
                 context=catch(
                     context, status=FAILED, updated={"errors": to_dict(e)}
                 ),
-                extras=self.extras,
             )
         finally:
             context["info"].update(

src/ddeutil/workflow/workflow.py

@@ -19,11 +19,12 @@
 Constants:
     NORMAL: Normal release execution
     RERUN: Re-execution of failed workflows
-    EVENT: Event-triggered execution
+    DRYRUN: Dryrun execution for testing workflow loop.
     FORCE: Force execution regardless of conditions
 """
 import copy
 import time
+import traceback
 from concurrent.futures import (
     Future,
     ThreadPoolExecutor,
@@ -50,6 +51,7 @@
     WorkflowError,
     WorkflowSkipError,
     WorkflowTimeoutError,
+    to_dict,
 )
 from .event import Event
 from .job import Job
@@ -667,22 +669,9 @@ def process(
 
                 if check == FAILED:  # pragma: no cov
                     pop_sys_extras(self.extras)
-                    return Result(
-                        run_id=run_id,
-                        parent_run_id=parent_run_id,
-                        status=FAILED,
-                        context=catch(
-                            context,
-                            status=FAILED,
-                            updated={
-                                "status": FAILED,
-                                "errors": WorkflowError(
-                                    f"Validate job trigger rule was failed "
-                                    f"with {job.trigger_rule.value!r}."
-                                ).to_dict(),
-                            },
-                        ),
-                        extras=self.extras,
+                    raise WorkflowError(
+                        f"Validate job trigger rule was failed with "
+                        f"{job.trigger_rule.value!r}."
                     )
                 elif check == SKIP:  # pragma: no cov
                     trace.info(
@@ -1026,12 +1015,23 @@ def execute(
                 trace.error(f"⏭️ Skip: {e}", module="workflow")
                 updated = None
             else:
-                trace.error(f"📢 Workflow Failed: {e}", module="workflow")
+                trace.error(f"📢 Workflow Failed:||{e}", module="workflow")
 
             st: Status = get_status_from_error(e)
             return Result.from_trace(trace).catch(
                 status=st, context=catch(context, status=st, updated=updated)
             )
+        except Exception as e:
+            trace.error(
+                f"💥 Error Failed:||🚨 {traceback.format_exc()}||",
+                module="workflow",
+            )
+            return Result.from_trace(trace).catch(
+                status=FAILED,
+                context=catch(
+                    context, status=FAILED, updated={"errors": to_dict(e)}
+                ),
+            )
         finally:
             context["info"].update(
                 {