♻️ clean: remove construct method on result object.

Author: korawica

src/ddeutil/workflow/job.py

@@ -1136,7 +1136,6 @@ def self_hosted_execute(
     params: DictData,
     *,
     run_id: StrOrNone = None,
-    parent_run_id: StrOrNone = None,
     event: Optional[Event] = None,
 ) -> Result:  # pragma: no cov
     """Self-Hosted job execution with passing dynamic parameters from the
@@ -1146,30 +1145,35 @@ def self_hosted_execute(
     :param job: (Job) A job model that want to execute.
     :param params: (DictData) A parameter data.
     :param run_id: (str) A job running ID.
-    :param parent_run_id: (str) A parent workflow running ID.
     :param event: (Event) An Event manager instance that use to cancel this
         execution if it forces stopped by parent execution.
 
     :rtype: Result
     """
-    result: Result = Result.construct_with_rs_or_id(
-        run_id=run_id,
-        parent_run_id=parent_run_id,
-        id_logic=(job.id or "EMPTY"),
-        extras=job.extras,
+    parent_run_id: StrOrNone = run_id
+    run_id: str = gen_id((job.id or "EMPTY"), unique=True)
+    trace: Trace = get_trace(
+        run_id, parent_run_id=parent_run_id, extras=job.extras
     )
-
-    result.trace.info("[JOB]: Start self-hosted executor.")
+    context: DictData = {"status": WAIT}
+    trace.info("[JOB]: Start self-hosted executor.")
 
     if event and event.is_set():
-        return result.catch(
+        return Result(
+            run_id=run_id,
+            parent_run_id=parent_run_id,
             status=CANCEL,
-            context={
-                "errors": JobCancelError(
-                    "Execution was canceled from the event before start "
-                    "self-hosted execution."
-                ).to_dict()
-            },
+            context=catch(
+                context,
+                status=CANCEL,
+                updated={
+                    "errors": JobCancelError(
+                        "Execution was canceled from the event before start "
+                        "self-hosted execution."
+                    ).to_dict()
+                },
+            ),
+            extras=job.extras,
         )
 
     import requests
@@ -1181,27 +1185,41 @@ def self_hosted_execute(
             data={
                 "job": job.model_dump(),
                 "params": params,
-                "result": result.__dict__,
+                "run_id": parent_run_id,
+                "extras": job.extras,
             },
         )
     except requests.exceptions.RequestException as e:
-        return result.catch(status=FAILED, context={"errors": to_dict(e)})
+        return Result(
+            run_id=run_id,
+            parent_run_id=parent_run_id,
+            status=FAILED,
+            context=catch(
+                context, status=FAILED, updated={"errors": to_dict(e)}
+            ),
+            extras=job.extras,
+        )
 
     if resp.status_code != 200:
         raise JobError(
             f"Job execution got error response from self-hosted: "
             f"{job.runs_on.args.host!r}"
         )
 
-    return result.catch(status=SUCCESS)
+    return Result(
+        run_id=run_id,
+        parent_run_id=parent_run_id,
+        status=SUCCESS,
+        context=catch(context, status=SUCCESS),
+        extras=job.extras,
+    )
 
 
 def azure_batch_execute(
     job: Job,
     params: DictData,
     *,
     run_id: StrOrNone = None,
-    parent_run_id: StrOrNone = None,
     event: Optional[Event] = None,
 ) -> Result:  # pragma: no cov
     """Azure Batch job execution that will run all job's stages on the Azure
@@ -1225,32 +1243,43 @@ def azure_batch_execute(
     :param job:
     :param params:
     :param run_id:
-    :param parent_run_id:
     :param event:
 
     :rtype: Result
     """
-    result: Result = Result.construct_with_rs_or_id(
-        run_id=run_id,
-        parent_run_id=parent_run_id,
-        id_logic=(job.id or "EMPTY"),
-        extras=job.extras,
+    parent_run_id: StrOrNone = run_id
+    run_id: str = gen_id((job.id or "EMPTY"), unique=True)
+    trace: Trace = get_trace(
+        run_id, parent_run_id=parent_run_id, extras=job.extras
     )
-
-    result.trace.info("[JOB]: Start Azure Batch executor.")
+    context: DictData = {"status": WAIT}
+    trace.info("[JOB]: Start Azure Batch executor.")
 
     if event and event.is_set():
-        return result.catch(
+        return Result(
+            run_id=run_id,
+            parent_run_id=parent_run_id,
             status=CANCEL,
-            context={
-                "errors": JobCancelError(
-                    "Execution was canceled from the event before start "
-                    "azure-batch execution."
-                ).to_dict()
-            },
+            context=catch(
+                context,
+                status=CANCEL,
+                updated={
+                    "errors": JobCancelError(
+                        "Execution was canceled from the event before start "
+                        "self-hosted execution."
+                    ).to_dict()
+                },
+            ),
+            extras=job.extras,
         )
     print(params)
-    return result.catch(status=SUCCESS)
+    return Result(
+        run_id=run_id,
+        parent_run_id=parent_run_id,
+        status=SUCCESS,
+        context=catch(context, status=SUCCESS),
+        extras=job.extras,
+    )
 
 
 def docker_execution(
@@ -1268,24 +1297,36 @@ def docker_execution(
         - Install this workflow package
         - Start push job to run to target Docker container.
     """
-    result: Result = Result.construct_with_rs_or_id(
-        run_id=run_id,
-        parent_run_id=parent_run_id,
-        id_logic=(job.id or "EMPTY"),
-        extras=job.extras,
+    parent_run_id: StrOrNone = run_id
+    run_id: str = gen_id((job.id or "EMPTY"), unique=True)
+    trace: Trace = get_trace(
+        run_id, parent_run_id=parent_run_id, extras=job.extras
     )
-
-    result.trace.info("[JOB]: Start Docker executor.")
+    context: DictData = {"status": WAIT}
+    trace.info("[JOB]: Start Docker executor.")
 
     if event and event.is_set():
-        return result.catch(
+        return Result(
+            run_id=run_id,
+            parent_run_id=parent_run_id,
             status=CANCEL,
-            context={
-                "errors": JobCancelError(
-                    "Execution was canceled from the event before start "
-                    "start docker execution."
-                ).to_dict()
-            },
+            context=catch(
+                context,
+                status=CANCEL,
+                updated={
+                    "errors": JobCancelError(
+                        "Execution was canceled from the event before start "
+                        "self-hosted execution."
+                    ).to_dict()
+                },
+            ),
+            extras=job.extras,
         )
     print(params)
-    return result.catch(status=SUCCESS)
+    return Result(
+        run_id=run_id,
+        parent_run_id=parent_run_id,
+        status=SUCCESS,
+        context=catch(context, status=SUCCESS),
+        extras=job.extras,
+    )

src/ddeutil/workflow/result.py

@@ -43,7 +43,7 @@
 from .__types import DictData
 from .audits import Trace, get_trace
 from .errors import ResultError
-from .utils import default_gen_id, gen_id, get_dt_ntz_now
+from .utils import default_gen_id, get_dt_ntz_now
 
 
 class Status(str, Enum):
@@ -194,60 +194,6 @@ class Result:
     trace: Optional[Trace] = field(default=None, compare=False, repr=False)
     extras: DictData = field(default_factory=dict, compare=False, repr=False)
 
-    @classmethod
-    def construct_with_id(
-        cls,
-        run_id: Optional[str] = None,
-        parent_run_id: Optional[str] = None,
-        id_logic: Optional[str] = None,
-        *,
-        extras: DictData | None = None,
-    ):
-        return cls(
-            run_id=(run_id or gen_id(id_logic or "", unique=True)),
-            parent_run_id=parent_run_id,
-            ts=get_dt_ntz_now(),
-            extras=(extras or {}),
-        )
-
-    @classmethod
-    def construct_with_rs_or_id(
-        cls,
-        result: Optional[Result] = None,
-        run_id: Optional[str] = None,
-        parent_run_id: Optional[str] = None,
-        id_logic: Optional[str] = None,
-        *,
-        extras: DictData | None = None,
-    ) -> Self:
-        """Create the Result object or set parent running id if passing Result
-        object.
-
-        Args:
-            result: A Result instance.
-            run_id: A running ID.
-            parent_run_id: A parent running ID.
-            id_logic: A logic function that use to generate a running ID.
-            extras: An extra parameter that want to override the core config.
-
-        Returns:
-            Self: The constructed or updated Result instance.
-        """
-        if result is None:
-            return cls(
-                run_id=(run_id or gen_id(id_logic or "", unique=True)),
-                parent_run_id=parent_run_id,
-                ts=get_dt_ntz_now(),
-                extras=(extras or {}),
-            )
-        elif parent_run_id:
-            result.set_parent_run_id(parent_run_id)
-
-        if extras is not None:
-            result.extras.update(extras)
-
-        return result
-
     @model_validator(mode="after")
     def __prepare_trace(self) -> Self:
         """Prepare trace field that want to pass after its initialize step.

tests/test_result.py

@@ -38,23 +38,6 @@ def test_result_default():
     assert rs.alive_time() > 0
 
 
-def test_result_construct_with_rs_or_id():
-    rs = Result.construct_with_rs_or_id(run_id="foo", extras={"test": "value"})
-    assert rs.run_id == "foo"
-    assert rs.parent_run_id is None
-    assert rs.extras == {"test": "value"}
-
-    rs = Result.construct_with_rs_or_id(
-        run_id="foo",
-        parent_run_id="baz",
-        result=Result(run_id="bar"),
-    )
-
-    assert rs.run_id != "foo"
-    assert rs.run_id == "bar"
-    assert rs.parent_run_id == "baz"
-
-
 def test_result_context():
     rs: Result = Result(context={"params": {"source": "src", "target": "tgt"}})
     rs.context.update({"additional-key": "new-value-to-add"})