🎨 format: change path type of audit model.

Author: korawica

docs/examples/conf/02-ml-model-training.yml

@@ -1,7 +1,12 @@
 name: "ml-model-training"
 type: "Workflow"
-description: "Machine learning model training pipeline with hyperparameter optimization and model evaluation"
-
+description: |
+  Machine learning model training pipeline with hyperparameter optimization
+  and model evaluation
+on:
+  schedule:
+    - cronjob: "0 0 * * 0"  # Weekly on Sunday at midnight
+      timezone: "UTC"
 params:
   dataset_path:
     type: str
@@ -10,7 +15,10 @@ params:
 
   model_type:
     type: choice
-    options: ["random_forest", "xgboost", "neural_network"]
+    options:
+      - "random_forest"
+      - "xgboost"
+      - "neural_network"
     desc: "Type of model to train"
 
   experiment_name:
@@ -370,8 +378,3 @@ jobs:
           channel: "#ml-deployments"
           message: "Model ${{ params.experiment_name }} v1 deployed successfully"
           color: "good"
-
-on:
-  schedule:
-    - cronjob: "0 0 * * 0"  # Weekly on Sunday at midnight
-      timezone: "UTC"

src/ddeutil/workflow/__init__.py

@@ -50,12 +50,12 @@
 from .__cron import CronRunner
 from .__types import DictData, DictStr, Matrix, Re, TupleStr
 from .audits import (
-    EVENT,
+    DRYRUN,
     FORCE,
     NORMAL,
     RERUN,
     Audit,
-    FileAudit,
+    LocalFileAudit,
     get_audit,
 )
 from .conf import (

src/ddeutil/workflow/audits.py

@@ -74,31 +74,38 @@ class ReleaseType(str, Enum):
     Attributes:
         NORMAL: Standard workflow release execution
         RERUN: Re-execution of previously failed workflow
-        EVENT: Event-triggered workflow execution
+        DRYRUN: Dry-execution workflow
         FORCE: Forced execution bypassing normal conditions
     """
 
     NORMAL = "normal"
     RERUN = "rerun"
-    EVENT = "event"
     FORCE = "force"
+    DRYRUN = "dryrun"
 
 
 NORMAL = ReleaseType.NORMAL
 RERUN = ReleaseType.RERUN
-EVENT = ReleaseType.EVENT
+DRYRUN = ReleaseType.DRYRUN
 FORCE = ReleaseType.FORCE
 
 
 class AuditData(BaseModel):
-    """Audit Data model."""
+    """Audit Data model that use to be the core data for any Audit model manage
+    logging at the target pointer system or service like file-system, sqlite
+    database, etc.
+    """
 
     model_config = ConfigDict(use_enum_values=True)
 
     name: str = Field(description="A workflow name.")
     release: datetime = Field(description="A release datetime.")
     type: ReleaseType = Field(
-        default=NORMAL, description="A running type before logging."
+        default=NORMAL,
+        description=(
+            "An execution type that should be value in ('normal', 'rerun', "
+            "'force', 'dryrun')."
+        ),
     )
     context: DictData = Field(
         default_factory=dict,
@@ -121,18 +128,17 @@ class BaseAudit(BaseModel, ABC):
     for logging subclasses like file, sqlite, etc.
     """
 
-    type: str
+    type: Literal["base"] = "base"
+    logging_name: str = "ddeutil.workflow"
     extras: DictData = Field(
         default_factory=dict,
         description="An extras parameter that want to override core config",
     )
 
     @field_validator("extras", mode="before")
-    def validate_extras(cls, v: Any) -> DictData:
+    def __prepare_extras(cls, v: Any) -> Any:
         """Validate extras field to ensure it's a dictionary."""
-        if v is None:
-            return {}
-        return v
+        return {} if v is None else v
 
     @model_validator(mode="after")
     def __model_action(self) -> Self:
@@ -148,7 +154,7 @@ def __model_action(self) -> Self:
             self.do_before()
 
         # NOTE: Start setting log config in this line with cache.
-        set_logging("ddeutil.workflow")
+        set_logging(self.logging_name)
         return self
 
     @abstractmethod
@@ -248,27 +254,33 @@ def save(
         raise NotImplementedError("Audit should implement `save` method.")
 
 
-class FileAudit(BaseAudit):
+class LocalFileAudit(BaseAudit):
     """File Audit Pydantic Model for saving log data from workflow execution.
 
     This class inherits from BaseAudit and implements file-based storage
     for audit logs. It saves workflow execution results to JSON files
     in a structured directory hierarchy.
 
     Attributes:
-        filename_fmt: Class variable defining the filename format for audit files.
+        file_fmt: Class variable defining the filename format for audit log.
+        file_release_fmt: Class variable defining the filename format for audit
+            release log.
     """
 
-    filename_fmt: ClassVar[str] = (
-        "workflow={name}/release={release:%Y%m%d%H%M%S}"
-    )
+    file_fmt: ClassVar[str] = "workflow={name}"
+    file_release_fmt: ClassVar[str] = "release={release:%Y%m%d%H%M%S}"
 
     type: Literal["file"] = "file"
-    path: str = Field(
-        default="./audits",
+    path: Path = Field(
+        default=Path("./audits"),
         description="A file path that use to manage audit logs.",
     )
 
+    @field_validator("path", mode="before", json_schema_input_type=str)
+    def __prepare_path(cls, data: Any) -> Any:
+        """Prepare path that passing with string to Path instance."""
+        return Path(data) if isinstance(data, str) else data
+
     def do_before(self) -> None:
         """Create directory of release before saving log file.
 
@@ -278,7 +290,10 @@ def do_before(self) -> None:
         Path(self.path).mkdir(parents=True, exist_ok=True)
 
     def find_audits(
-        self, name: str, *, extras: Optional[DictData] = None
+        self,
+        name: str,
+        *,
+        extras: Optional[DictData] = None,
     ) -> Iterator[AuditData]:
         """Generate audit data found from logs path for a specific workflow name.
 
@@ -292,7 +307,7 @@ def find_audits(
         Raises:
             FileNotFoundError: If the workflow directory does not exist.
         """
-        pointer: Path = Path(self.path) / f"workflow={name}"
+        pointer: Path = self.path / self.file_fmt.format(name=name)
         if not pointer.exists():
             raise FileNotFoundError(f"Pointer: {pointer.absolute()}.")
 
@@ -325,7 +340,7 @@ def find_audit_with_release(
             ValueError: If no releases found when release is None.
         """
         if release is None:
-            pointer: Path = Path(self.path) / f"workflow={name}"
+            pointer: Path = self.path / self.file_fmt.format(name=name)
             if not pointer.exists():
                 raise FileNotFoundError(f"Pointer: {pointer.absolute()}.")
 
@@ -382,8 +397,10 @@ def pointer(self, data: AuditData) -> Path:
         Returns:
             Path: The directory path for the current workflow and release.
         """
-        return Path(self.path) / self.filename_fmt.format(
-            name=data.name, release=data.release
+        return (
+            self.path
+            / self.file_fmt.format(**data.model_dump(by_alias=True))
+            / self.file_release_fmt.format(**data.model_dump(by_alias=True))
         )
 
     def save(self, data: Any, excluded: Optional[list[str]] = None) -> Self:
@@ -459,19 +476,19 @@ def cleanup(self, max_age_days: int = 180) -> int:  # pragma: no cov
         return cleaned_count
 
 
-class SQLiteAudit(BaseAudit):  # pragma: no cov
+class LocalSQLiteAudit(BaseAudit):  # pragma: no cov
     """SQLite Audit model for database-based audit storage.
 
     This class inherits from BaseAudit and implements SQLite database storage
     for audit logs with compression support.
 
     Attributes:
         table_name: Class variable defining the database table name.
-        schemas: Class variable defining the database schema.
+        ddl: Class variable defining the database schema.
     """
 
     table_name: ClassVar[str] = "audits"
-    schemas: ClassVar[
+    ddl: ClassVar[
         str
     ] = """
         CREATE TABLE IF NOT EXISTS audits (
@@ -489,22 +506,21 @@ class SQLiteAudit(BaseAudit):  # pragma: no cov
         """
 
     type: Literal["sqlite"] = "sqlite"
-    path: str
+    path: Path = Field(
+        default=Path("./audits.db"),
+        description="A SQLite filepath.",
+    )
 
-    def _ensure_table_exists(self) -> None:
+    def do_before(self) -> None:
         """Ensure the audit table exists in the database."""
-        audit_url = dynamic("audit_url", extras=self.extras)
-        if audit_url is None or not audit_url.path:
+        if self.path.is_dir():
             raise ValueError(
-                "SQLite audit_url must specify a database file path"
+                "SQLite path must specify a database file path not dir."
             )
 
-        audit_url_parse: ParseResult = urlparse(audit_url)
-        db_path = Path(audit_url_parse.path)
-        db_path.parent.mkdir(parents=True, exist_ok=True)
-
-        with sqlite3.connect(db_path) as conn:
-            conn.execute(self.schemas)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with sqlite3.connect(self.path) as conn:
+            conn.execute(self.ddl)
             conn.commit()
 
     def is_pointed(
@@ -771,24 +787,31 @@ def cleanup(self, max_age_days: int = 180) -> int:
             return cursor.rowcount
 
 
+class PostgresAudit(BaseAudit, ABC): ...  # pragma: no cov
+
+
 Audit = Annotated[
     Union[
-        FileAudit,
-        SQLiteAudit,
+        LocalFileAudit,
+        LocalSQLiteAudit,
     ],
     Field(discriminator="type"),
 ]
 
 
-def get_audit(extras: Optional[DictData] = None) -> Audit:  # pragma: no cov
+def get_audit(
+    audit_conf: Optional[DictData] = None,
+    extras: Optional[DictData] = None,
+) -> Audit:  # pragma: no cov
     """Get an audit model dynamically based on the config audit path value.
 
     Args:
+        audit_conf (DictData):
         extras: Optional extra parameters to override the core config.
 
     Returns:
         Audit: The appropriate audit model class based on configuration.
     """
-    audit_conf = dynamic("audit_conf", extras=extras)
+    audit_conf = dynamic("audit_conf", f=audit_conf, extras=extras)
     model = TypeAdapter(Audit).validate_python(audit_conf | {"extras": extras})
     return model

src/ddeutil/workflow/traces.py

@@ -239,7 +239,7 @@ class Metadata(BaseModel):  # pragma: no cov
         default=None, description="Environment (dev, staging, prod)."
     )
 
-    # System context
+    # NOTE: System context
     hostname: Optional[str] = Field(
         default=None, description="Hostname where workflow is running."
     )
@@ -253,7 +253,7 @@ class Metadata(BaseModel):  # pragma: no cov
         default=None, description="Workflow package version."
     )
 
-    # Custom metadata
+    # NOTE: Custom metadata
     tags: Optional[list[str]] = Field(
         default_factory=list, description="Custom tags for categorization."
     )
@@ -320,6 +320,8 @@ def make(
         import socket
         import sys
 
+        from .__about__ import __version__
+
         frame: Optional[FrameType] = currentframe()
         if frame is None:
             raise ValueError("Cannot get current frame")
@@ -384,7 +386,7 @@ def make(
             hostname=hostname,
             ip_address=ip_address,
             python_version=python_version,
-            package_version=extras_data.get("package_version"),
+            package_version=__version__,
             # NOTE: Custom metadata
             tags=extras_data.get("tags", []),
             metadata=extras_data.get("metadata", {}),
@@ -2046,7 +2048,7 @@ def get_trace(
     Args:
         run_id (str): A running ID.
         parent_run_id (str | None, default None): A parent running ID.
-        handlers:
+        handlers (list):
         extras: An extra parameter that want to override the core
             config values.
         auto_pre_process (bool, default False)
@@ -2057,7 +2059,7 @@ def get_trace(
     handlers: list[DictData] = dynamic(
         "trace_handlers", f=handlers, extras=extras
     )
-    trace = Trace.model_validate(
+    trace: Trace = Trace.model_validate(
         {
             "run_id": run_id,
             "parent_run_id": parent_run_id,

src/ddeutil/workflow/utils.py

@@ -8,26 +8,6 @@
 This module provides essential utility functions used throughout the workflow
 system for ID generation, datetime handling, string processing, template
 operations, and other common tasks.
-
-Functions:
-    to_train: Convert camel case strings to train case format
-    prepare_newline: Format messages with multiple newlines
-    replace_sec: Replace seconds and microseconds in datetime objects
-    clear_tz: Clear timezone info from datetime objects
-    get_dt_now: Get current datetime with timezone
-    get_d_now: Get current date
-    get_diff_sec: Calculate time difference in seconds
-    reach_next_minute: Check if datetime reaches next minute
-    wait_until_next_minute: Wait until next minute
-    delay: Add random delay to execution
-    gen_id: Generate unique identifiers for workflow components
-    default_gen_id: Generate default running ID
-    make_exec: Make files executable
-    filter_func: Filter function objects from data structures
-    cross_product: Generate cross product of matrix values
-    cut_id: Cut running ID to specified length
-    dump_all: Serialize nested BaseModel objects to dictionaries
-    obj_name: Get object name or class name
 """
 from __future__ import annotations
 
@@ -327,6 +307,8 @@ def cut_id(run_id: str, *, num: int = 8) -> str:
     Example:
         >>> cut_id(run_id='20240101081330000000T1354680202')
         '202401010813680202'
+        >>> cut_id(run_id='20240101081330000000T1354680202')
+        '54680202'
 
     Args:
         run_id: A running ID to cut.