Feature/722 nox session check workflow orchestrator

.github/workflows/fast-tests-extension.yml

@@ -16,7 +16,7 @@ jobs:
 
       - name: Set up Python & Poetry Environment
         id: set-up-python-and-poetry-environment
-        uses: exasol/python-toolbox/.github/actions/python-environment@v7
+        uses: exasol/python-toolbox/.github/actions/python-environment@v8
         with:
           python-version: "3.10"
           poetry-version: "2.3.0"

.github/workflows/slow-checks.yml

@@ -11,7 +11,7 @@ jobs:
       contents: read
 
   run-integration-tests:
-    name: Run Integration Tests (Python-${{ matrix.python-version }}, Exasol-${{ matrix.exasol-version}})
+    name: Run Integration Tests (Python-${{ matrix.python-version }})
     needs:
       - build-matrix
     runs-on: "ubuntu-24.04"
@@ -29,7 +29,7 @@ jobs:
 
       - name: Set up Python & Poetry Environment
         id: set-up-python-and-poetry-environment
-        uses: exasol/python-toolbox/.github/actions/python-environment@v7
+        uses: exasol/python-toolbox/.github/actions/python-environment@v8
         with:
           python-version: ${{ matrix.python-version }}
           poetry-version: "2.3.0"

doc/changes/unreleased.md

@@ -9,3 +9,7 @@
 ## Feature
 
 * #722: Added check in `workflow:generate` to compare the generated and existing content before writing out
+
+## Refactoring
+
+* #722: Modified `workflow:generate` backend function to class `WorkflowOrchestrator`

exasol/toolbox/nox/_workflow.py

@@ -5,9 +5,9 @@
 import nox
 from nox import Session
 
-from exasol.toolbox.util.workflows.workflow import (
+from exasol.toolbox.util.workflows.workflow_orchestrator import (
     WORKFLOW_CHOICES,
-    update_workflow,
+    WorkflowOrchestrator,
 )
 from noxconfig import PROJECT_CONFIG
 
@@ -37,4 +37,7 @@ def generate_workflow(session: Session) -> None:
 
     PROJECT_CONFIG.github_workflow_directory.mkdir(parents=True, exist_ok=True)
 
-    update_workflow(workflow_choice=args.workflow_choice, config=PROJECT_CONFIG)
+    WorkflowOrchestrator(
+        workflow_choice=args.workflow_choice,
+        config=PROJECT_CONFIG,
+    ).generate_workflows()

exasol/toolbox/util/workflows/workflow.py

@@ -1,10 +1,7 @@
 import difflib
-from collections.abc import Mapping
 from pathlib import Path
 from typing import (
-    Annotated,
     Any,
-    Final,
 )
 
 from pydantic import (
@@ -15,28 +12,15 @@
     bound_contextvars,
 )
 
-from exasol.toolbox.config import BaseConfig
 from exasol.toolbox.util.workflows import logger
 from exasol.toolbox.util.workflows.exceptions import (
-    InvalidWorkflowPatcherEntryError,
-    NotMaintainedWorkflowError,
     YamlError,
     YamlKeyError,
 )
 from exasol.toolbox.util.workflows.patch_workflow import (
     WorkflowCommentedMap,
-    WorkflowPatcher,
 )
 from exasol.toolbox.util.workflows.process_template import WorkflowRenderer
-from exasol.toolbox.util.workflows.templates import (
-    WORKFLOW_TEMPLATE_OPTIONS,
-    validate_workflow_name,
-)
-
-ALL: Final[str] = "all"
-WORKFLOW_CHOICES: Final[list[str]] = [ALL, *WORKFLOW_TEMPLATE_OPTIONS.keys()]
-
-WorkflowChoice = Annotated[str, f"Should be a value from {WORKFLOW_CHOICES}"]
 
 
 class Workflow(BaseModel):
@@ -78,9 +62,10 @@ def load_from_template(
                 raise ValueError(f"Error rendering file: {template_path}") from ex
 
     def compare_to_file(self) -> str:
-        existing_content = (
-            self.output_path.read_text().strip() if self.output_path.exists() else ""
-        )
+        existing_content = ""
+        if self.output_path.is_file():
+            existing_content = self.output_path.read_text().strip()
+
         generated_content = self.content.strip()
 
         diff = difflib.unified_diff(
@@ -98,59 +83,3 @@ def write_to_file(self) -> None:
             return
         logger.info("Write workflow file %s", self.output_path.name)
         self.output_path.write_text(self.content + "\n")
-
-
-def _select_workflow_template(workflow_name: WorkflowChoice) -> Mapping[str, Path]:
-    """
-    Returns a mapping of workflow names to paths. Can be a single item or all workflow
-    templates.
-    """
-    if workflow_name == ALL:
-        return WORKFLOW_TEMPLATE_OPTIONS
-    return {workflow_name: WORKFLOW_TEMPLATE_OPTIONS[workflow_name]}
-
-
-def update_workflow(workflow_choice: WorkflowChoice, config: BaseConfig) -> None:
-    """
-    Updates a selected workflow or all workflows.
-    """
-    workflow_dict = _select_workflow_template(workflow_choice)
-    logger.info(f"Selected workflow(s) to update: {list(workflow_dict.keys())}")
-
-    workflow_patcher = None
-    if config.github_workflow_patcher_yaml:
-        workflow_patcher = WorkflowPatcher(
-            github_template_dict=config.github_template_dict,
-            file_path=config.github_workflow_patcher_yaml,
-        )
-
-    is_new_project = not any(config.github_workflow_directory.glob("*.yml"))
-    for workflow_name in workflow_dict:
-        patch_yaml = None
-        if workflow_patcher:
-            patch_yaml = workflow_patcher.extract_by_workflow(
-                workflow_name=workflow_name
-            )
-
-        try:
-            validate_workflow_name(workflow_name)
-        except NotMaintainedWorkflowError:
-            if not is_new_project:
-                logger.debug(
-                    "Skipping not-maintained workflow in older project: %s",
-                    workflow_name,
-                )
-                continue
-
-        try:
-            workflow = Workflow.load_from_template(
-                template_path=workflow_dict[workflow_name],
-                output_directory=config.github_workflow_directory,
-                github_template_dict=config.github_template_dict,
-                patch_yaml=patch_yaml,
-            )
-            workflow.write_to_file()
-        except YamlKeyError as ex:
-            raise InvalidWorkflowPatcherEntryError(
-                file_path=config.github_workflow_patcher_yaml, entry=ex.entry  # type: ignore
-            ) from ex

exasol/toolbox/util/workflows/workflow_orchestrator.py

@@ -0,0 +1,131 @@
+from __future__ import annotations
+
+from collections.abc import (
+    Iterator,
+    Mapping,
+)
+from functools import cached_property
+from pathlib import Path
+from typing import (
+    Annotated,
+    Final,
+)
+
+from pydantic import BaseModel
+
+from exasol.toolbox.config import BaseConfig
+from exasol.toolbox.util.workflows import logger
+from exasol.toolbox.util.workflows.exceptions import (
+    InvalidWorkflowPatcherEntryError,
+    NotMaintainedWorkflowError,
+    YamlKeyError,
+)
+from exasol.toolbox.util.workflows.patch_workflow import (
+    WorkflowCommentedMap,
+    WorkflowPatcher,
+)
+from exasol.toolbox.util.workflows.templates import (
+    WORKFLOW_TEMPLATE_OPTIONS,
+    validate_workflow_name,
+)
+from exasol.toolbox.util.workflows.workflow import Workflow
+
+ALL: Final[str] = "all"
+WorkflowChoice = Annotated[
+    str, f"Should be a value from {[ALL, *WORKFLOW_TEMPLATE_OPTIONS.keys()]}"
+]
+WORKFLOW_CHOICES: Final[list[str]] = [ALL, *WORKFLOW_TEMPLATE_OPTIONS.keys()]
+
+
+class WorkflowOrchestrator(BaseModel):
+    """Orchestrate workflow rendering, comparison, and writing."""
+
+    workflow_choice: WorkflowChoice
+    config: BaseConfig
+
+    @cached_property
+    def templates(self) -> Mapping[str, Path]:
+        """
+        A mapping of workflow templates names to paths. This can be a single
+        item or all workflow templates.
+        """
+        if self.workflow_choice == ALL:
+            return WORKFLOW_TEMPLATE_OPTIONS
+        return {self.workflow_choice: WORKFLOW_TEMPLATE_OPTIONS[self.workflow_choice]}
+
+    @cached_property
+    def workflow_patcher(self) -> WorkflowPatcher | None:
+        if not self.config.github_workflow_patcher_yaml:
+            return None
+        return WorkflowPatcher(
+            github_template_dict=self.config.github_template_dict,
+            file_path=self.config.github_workflow_patcher_yaml,
+        )
+
+    def _extract_workflow_patch(
+        self, workflow_name: str
+    ) -> WorkflowCommentedMap | None:
+        """
+        Return the patch data for a workflow, or ``None`` if no patcher is configured.
+        """
+        if self.workflow_patcher is None:
+            return None
+        return self.workflow_patcher.extract_by_workflow(workflow_name=workflow_name)
+
+    def _is_new_project(self) -> bool:
+        """
+        A project is considered new if no YML files are present in the GitHub directory.
+        """
+        return not any(self.config.github_workflow_directory.glob("*.yml"))
+
+    def _iter_workflows(self) -> Iterator[Workflow]:
+        logger.info(f"Selected workflow(s) to update: {list(self.templates.keys())}")
+        is_new_project = self._is_new_project()
+        for workflow_name, template_path in self.templates.items():
+            patch_yaml = self._extract_workflow_patch(workflow_name=workflow_name)
+
+            if self._skip_workflow(workflow_name, is_new_project):
+                continue
+
+            yield self._load_workflow(
+                template_path=template_path, patch_yaml=patch_yaml
+            )
+
+    def _load_workflow(
+        self, template_path: Path, patch_yaml: WorkflowCommentedMap | None
+    ):
+        try:
+            return Workflow.load_from_template(
+                template_path=template_path,
+                output_directory=self.config.github_workflow_directory,
+                github_template_dict=self.config.github_template_dict,
+                patch_yaml=patch_yaml,
+            )
+        except YamlKeyError as ex:
+            raise InvalidWorkflowPatcherEntryError(
+                file_path=self.config.github_workflow_patcher_yaml,  # type: ignore
+                entry=ex.entry,
+            ) from ex
+
+    def _skip_workflow(self, workflow_name: str, is_new_project: bool) -> bool:
+        """
+        Return ``True`` if the workflow should be skipped because it is not maintained
+        by the PTB, otherwise return ``False``.
+        """
+        try:
+            validate_workflow_name(workflow_name)
+        except NotMaintainedWorkflowError:
+            if not is_new_project:
+                logger.debug(
+                    "Skipping not-maintained workflow in older project: %s",
+                    workflow_name,
+                )
+                return True
+        return False
+
+    def generate_workflows(self) -> None:
+        """
+        Render the selected workflows and write them to disk.
+        """
+        for workflow in self._iter_workflows():
+            workflow.write_to_file()