🎯 feat: add detail method for model dump with alias.

Author: korawica

src/ddeutil/workflow/job.py

@@ -48,10 +48,11 @@
 from functools import lru_cache
 from textwrap import dedent
 from threading import Event
-from typing import Annotated, Any, Literal, Optional, Union
+from typing import Annotated, Any, Optional, Union
 
 from ddeutil.core import freeze_args
 from pydantic import BaseModel, Discriminator, Field, SecretStr, Tag
+from pydantic.functional_serializers import field_serializer
 from pydantic.functional_validators import field_validator, model_validator
 from typing_extensions import Self
 
@@ -263,24 +264,20 @@ class BaseRunsOn(BaseModel):  # pragma: no cov
     object and override execute method.
     """
 
-    type: RunsOn = Field(description="A runs-on type.")
+    type: RunsOn = LOCAL
     args: DictData = Field(
         default_factory=dict,
-        alias="with",
         description=(
             "An argument that pass to the runs-on execution function. This "
             "args will override by this child-model with specific args model."
         ),
+        alias="with",
     )
 
 
 class OnLocal(BaseRunsOn):  # pragma: no cov
     """Runs-on local."""
 
-    type: Literal[RunsOn.LOCAL] = Field(
-        default=RunsOn.LOCAL, validate_default=True
-    )
-
 
 class SelfHostedArgs(BaseModel):
     """Self-Hosted arguments."""
@@ -292,9 +289,7 @@ class SelfHostedArgs(BaseModel):
 class OnSelfHosted(BaseRunsOn):  # pragma: no cov
     """Runs-on self-hosted."""
 
-    type: Literal[RunsOn.SELF_HOSTED] = Field(
-        default=RunsOn.SELF_HOSTED, validate_default=True
-    )
+    type: RunsOn = SELF_HOSTED
     args: SelfHostedArgs = Field(alias="with")
 
 
@@ -310,9 +305,7 @@ class AzBatchArgs(BaseModel):
 
 class OnAzBatch(BaseRunsOn):  # pragma: no cov
 
-    type: Literal[RunsOn.AZ_BATCH] = Field(
-        default=RunsOn.AZ_BATCH, validate_default=True
-    )
+    type: RunsOn = AZ_BATCH
     args: AzBatchArgs = Field(alias="with")
 
 
@@ -331,23 +324,21 @@ class DockerArgs(BaseModel):
 class OnDocker(BaseRunsOn):  # pragma: no cov
     """Runs-on Docker container."""
 
-    type: Literal[RunsOn.DOCKER] = Field(
-        default=RunsOn.DOCKER, validate_default=True
-    )
-    args: DockerArgs = Field(alias="with", default_factory=DockerArgs)
+    type: RunsOn = DOCKER
+    args: DockerArgs = Field(default_factory=DockerArgs, alias="with")
 
 
 def get_discriminator_runs_on(model: dict[str, Any]) -> RunsOn:
     """Get discriminator of the RunsOn models."""
     t: str = model.get("type")
-    return RunsOn(t) if t else RunsOn.LOCAL
+    return RunsOn(t) if t else LOCAL
 
 
 RunsOnModel = Annotated[
     Union[
-        Annotated[OnSelfHosted, Tag(RunsOn.SELF_HOSTED)],
-        Annotated[OnDocker, Tag(RunsOn.DOCKER)],
-        Annotated[OnLocal, Tag(RunsOn.LOCAL)],
+        Annotated[OnSelfHosted, Tag(SELF_HOSTED)],
+        Annotated[OnDocker, Tag(DOCKER)],
+        Annotated[OnLocal, Tag(LOCAL)],
     ],
     Discriminator(get_discriminator_runs_on),
 ]
@@ -490,6 +481,10 @@ def __validate_job_id__(self) -> Self:
 
         return self
 
+    @field_serializer("runs_on")
+    def __serialize_runs_on(self, value: RunsOnModel):
+        return value.model_dump(by_alias=True)
+
     def stage(self, stage_id: str) -> Stage:
         """Return stage instance that exists in this job via passing an input
         stage ID.

src/ddeutil/workflow/workflow.py

@@ -145,7 +145,7 @@ class Workflow(BaseModel):
         description="A parameters that need to use on this workflow.",
     )
     on: Event = Field(
-        default_factory=list,
+        default_factory=Event,
         description="An events for this workflow.",
     )
     jobs: dict[str, Job] = Field(
@@ -284,14 +284,18 @@ def __validate_jobs_need__(self) -> Self:
 
         return self
 
-    def md(self) -> str:  # pragma: no cov
+    def detail(self) -> DictData:  # pragma: no cov
+        """Return the detail of this workflow for generate markdown."""
+        return self.model_dump(by_alias=True)
+
+    def md(self, author: Optional[str] = None) -> str:  # pragma: no cov
         """Generate the markdown template."""
 
         def align_newline(value: str) -> str:
             return value.rstrip("\n").replace("\n", "\n                ")
 
         info: str = (
-            f"| Author: nobody "
+            f"| Author: {author or 'nobody'} "
             f"| created_at: `{self.created_at:%Y-%m-%d %H:%M:%S}` "
             f"| updated_at: `{self.updated_dt:%Y-%m-%d %H:%M:%S}` |\n"
             f"| --- | --- | --- |"
@@ -412,7 +416,7 @@ def validate_release(self, dt: datetime) -> datetime:
             dt = dt.replace(tzinfo=UTC)
 
         release: datetime = replace_sec(dt.astimezone(UTC))
-        if not self.on:
+        if not self.on.schedule:
             return release
 
         for on in self.on.schedule:

tests/test_workflow.py

@@ -427,6 +427,13 @@ def test_workflow_parameterize(test_path):
             workflow.parameterize({"foo": "bar"})
 
 
+def test_workflow_detail(test_path):
+    workflow = Workflow.from_conf(
+        "stream-workflow", path=test_path / "conf/example"
+    )
+    print(workflow.detail())
+
+
 def test_workflow_markdown(test_path):
     workflow = Workflow.from_conf(
         "stream-workflow", path=test_path / "conf/example"