Feature/719 add Nox session workflow:generate

doc/changes/unreleased.md

@@ -8,6 +8,7 @@
 * #712: Added basic logging to workflow processing
 * #714: Added logic to modify a workflow using the .workflow-patcher.yml
 * #717: Restricted workflow names in .workflow-patcher.yml to template workflow names
+* #719: Added nox session `workflow:generate` to generate/update workflows using the `.workflow-patcher.yml` (if desired)
 
 ## Documentation
 

exasol/toolbox/config.py

@@ -266,6 +266,11 @@ def version_filepath(self) -> Path:
         """
         return self.source_code_path / "version.py"
 
+    @computed_field  # type: ignore[misc]
+    @property
+    def github_workflow_directory(self) -> Path:
+        return self.root_path / ".github" / "workflows"
+
     @computed_field  # type: ignore[misc]
     @property
     def github_template_dict(self) -> dict[str, Any]:

exasol/toolbox/nox/_workflow.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+import argparse
+
+import nox
+from nox import Session
+
+from exasol.toolbox.util.workflows.workflow import (
+    WORKFLOW_CHOICES,
+    update_workflow,
+)
+from noxconfig import PROJECT_CONFIG
+
+
+def _create_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="nox -s workflow:generate",
+        usage="nox -s workflow:generate -- [-h]  <workflow_choice>",
+        formatter_class=argparse.ArgumentDefaultsHelpFormatter,
+    )
+
+    parser.add_argument(
+        "workflow_choice",
+        choices=WORKFLOW_CHOICES,
+        help="Select one workflow or 'all' to all workflows.",
+    )
+    return parser
+
+
+@nox.session(name="workflow:generate", python=False)
+def generate_workflow(session: Session) -> None:
+    """
+    Generate or update the specified GitHub workflow or all of them.
+    """
+    parser = _create_parser()
+    args = parser.parse_args(session.posargs)
+
+    PROJECT_CONFIG.github_workflow_directory.mkdir(parents=True, exist_ok=True)
+
+    update_workflow(workflow_choice=args.workflow_choice, config=PROJECT_CONFIG)

exasol/toolbox/nox/tasks.py

@@ -90,6 +90,7 @@ def check(session: Session) -> None:
 from exasol.toolbox.nox._package_version import version_check
 
 from exasol.toolbox.nox._package import package_check
+from exasol.toolbox.nox._workflow import generate_workflow
 
 # isort: on
 # fmt: on

exasol/toolbox/tools/workflow.py

@@ -76,7 +76,7 @@ def install_workflow(
         template=workflow, dest=dest, pkg=PKG, template_type=TEMPLATE_TYPE
     )
     warnings.warn(
-        "\033[31m`tbx workflow install` is deprecated; this will be replaced by a nox session after 2026-04-22\033[0m",
+        "\033[31m`tbx workflow install` is deprecated; this will be replaced by the Nox session `workflow:generate` after 2026-04-22\033[0m",
         category=FutureWarning,
         stacklevel=2,
     )
@@ -101,7 +101,7 @@ def update_workflow(
         template_type=TEMPLATE_TYPE,
     )
     warnings.warn(
-        "\033[31m`tbx workflow update` is deprecated; this will be replaced by a nox session after 2026-04-22\033[0m",
+        "\033[31m`tbx workflow update` is deprecated; this will be replaced by the Nox session `workflow:generate`  after 2026-04-22\033[0m",
         category=FutureWarning,
         stacklevel=2,
     )

exasol/toolbox/util/workflows/exceptions.py

@@ -1,15 +1,18 @@
+from collections.abc import Mapping
 from pathlib import Path
 
 
 class YamlError(Exception):
-    """Base exception for YAML errors."""
+    """
+    Base exception for YAML errors.
+    """
 
     message_template = "An error occurred with file: {file_path}"
 
-    def __init__(self, file_path: Path):
+    def __init__(self, file_path: Path, **kwargs):
         self.file_path = file_path
         # Format the template defined in the subclass
-        message = self.message_template.format(file_path=file_path)
+        message = self.message_template.format(file_path=file_path, **kwargs)
         super().__init__(message)
 
 
@@ -73,37 +76,31 @@ class YamlKeyError(Exception):
 
     message_template = "An error occurred with job: '{job_name}'"
 
-    def __init__(self, job_name: str):
-        self.job_name = job_name
-        # Format the template defined in the subclass
-        message = self.message_template.format(job_name=job_name)
-        super().__init__(message)
+    def __init__(self, **kwargs):
+        # Store all attributes dynamically (job_name, step_id, etc.)
+        for key, value in kwargs.items():
+            setattr(self, key, value)
+
+        self._data = kwargs
+        # Format the template using the passed-in arguments
+        super().__init__(self.message_template.format(**kwargs))
+
+    @property
+    def entry(self) -> Mapping[str, str]:
+        return self._data
 
 
-class YamlJobValueError(Exception):
+class YamlJobValueError(YamlKeyError):
     """
     Raised when a job cannot be found in a YAML file.
     """
 
     message_template = "Job '{job_name}' could not be found"
 
-    def __init__(self, job_name: str):
-        self.job_name = job_name
-        # Format the template defined in the subclass
-        message = self.message_template.format(job_name=job_name)
-        super().__init__(message)
-
 
 class YamlStepIdValueError(YamlKeyError):
     """
     Raised when a step_id associated with a specific job cannot be found in a YAML file.
     """
 
     message_template = "Step_id '{step_id}' not found in job '{job_name}'"
-
-    def __init__(self, step_id: str, job_name: str):
-        self.step_id = step_id
-        self.job_name = job_name
-
-        message = self.message_template.format(step_id=step_id, job_name=job_name)
-        super(YamlKeyError, self).__init__(message)

exasol/toolbox/util/workflows/patch_workflow.py

@@ -14,7 +14,9 @@
     ValidationError,
 )
 from ruamel.yaml import CommentedMap
+from structlog.contextvars import bound_contextvars
 
+from exasol.toolbox.util.workflows import logger
 from exasol.toolbox.util.workflows.exceptions import InvalidWorkflowPatcherYamlError
 from exasol.toolbox.util.workflows.render_yaml import YamlRenderer
 from exasol.toolbox.util.workflows.templates import WORKFLOW_TEMPLATE_OPTIONS
@@ -120,12 +122,15 @@ def content(self) -> CommentedMap:
         The loaded YAML content. It loads on first access and stays cached even though
         the class is frozen.
         """
-        loaded_yaml = self.get_yaml_dict()
-        try:
-            WorkflowPatcherConfig.model_validate(loaded_yaml)
-            return loaded_yaml
-        except ValidationError as ex:
-            raise InvalidWorkflowPatcherYamlError(file_path=self.file_path) from ex
+        with bound_contextvars(template_file_name=self.file_path.name):
+            logger.info("Load workflow patcher: %s", self.file_path.name)
+            loaded_yaml = self.get_yaml_dict()
+            try:
+                logger.debug("Validate workflow patcher with Pydantic")
+                WorkflowPatcherConfig.model_validate(loaded_yaml)
+                return loaded_yaml
+            except ValidationError as ex:
+                raise InvalidWorkflowPatcherYamlError(file_path=self.file_path) from ex
 
     def extract_by_workflow(self, workflow_name: str) -> WorkflowCommentedMap | None:
         """

exasol/toolbox/util/workflows/process_template.py

@@ -33,7 +33,7 @@ def render(self) -> str:
         workflow_dict = self.get_yaml_dict()
 
         if self.patch_yaml:
-            logger.debug("Modify workflow custom yaml")
+            logger.debug("Customize workflow with `patch_yaml`")
             workflow_modifier = WorkflowModifier(
                 workflow_dict=workflow_dict, patch_yaml=self.patch_yaml
             )

exasol/toolbox/util/workflows/workflow.py

@@ -1,5 +1,10 @@
+from collections.abc import Mapping
 from pathlib import Path
-from typing import Any
+from typing import (
+    Annotated,
+    Any,
+    Final,
+)
 
 from pydantic import (
     BaseModel,
@@ -9,9 +14,24 @@
     bound_contextvars,
 )
 
+from exasol.toolbox.config import BaseConfig
 from exasol.toolbox.util.workflows import logger
-from exasol.toolbox.util.workflows.exceptions import YamlError
+from exasol.toolbox.util.workflows.exceptions import (
+    InvalidWorkflowPatcherEntryError,
+    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
+
+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):
@@ -20,9 +40,14 @@ class Workflow(BaseModel):
     content: str
 
     @classmethod
-    def load_from_template(cls, file_path: Path, github_template_dict: dict[str, Any]):
+    def load_from_template(
+        cls,
+        file_path: Path,
+        github_template_dict: dict[str, Any],
+        patch_yaml: WorkflowCommentedMap | None = None,
+    ):
         with bound_contextvars(template_file_name=file_path.name):
-            logger.info("Load workflow from template")
+            logger.info("Load workflow template: %s", file_path.name)
 
             if not file_path.exists():
                 raise FileNotFoundError(file_path)
@@ -31,12 +56,61 @@ def load_from_template(cls, file_path: Path, github_template_dict: dict[str, Any
                 workflow_renderer = WorkflowRenderer(
                     github_template_dict=github_template_dict,
                     file_path=file_path,
-                    patch_yaml=None,
+                    patch_yaml=patch_yaml,
                 )
                 workflow = workflow_renderer.render()
                 return cls(content=workflow)
-            except YamlError as ex:
+            except (YamlError, YamlKeyError) as ex:
                 raise ex
             except Exception as ex:
                 # Wrap all other "non-special" exceptions
                 raise ValueError(f"Error rendering file: {file_path}") from ex
+
+    def write_to_file(self, file_path: Path) -> None:
+        logger.info("Write workflow file %s", file_path.name)
+        file_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,
+        )
+
+    for workflow_name in workflow_dict:
+        patch_yaml = None
+        if workflow_patcher:
+            patch_yaml = workflow_patcher.extract_by_workflow(
+                workflow_name=workflow_name
+            )
+
+        try:
+            workflow = Workflow.load_from_template(
+                file_path=workflow_dict[workflow_name],
+                github_template_dict=config.github_template_dict,
+                patch_yaml=patch_yaml,
+            )
+            file_path = config.github_workflow_directory / f"{workflow_name}.yml"
+            workflow.write_to_file(file_path=file_path)
+        except YamlKeyError as ex:
+            raise InvalidWorkflowPatcherEntryError(
+                file_path=config.github_workflow_patcher_yaml, entry=ex.entry  # type: ignore
+            ) from ex

test/unit/config_test.py

@@ -15,7 +15,7 @@
 
 class TestBaseConfig:
     @staticmethod
-    def test_works_as_defined(test_project_config_factory):
+    def test_works_as_defined(tmp_path, test_project_config_factory):
         config = test_project_config_factory()
 
         root_path = config.root_path
@@ -34,6 +34,7 @@ def test_works_as_defined(test_project_config_factory):
                 "dist",
                 "venv",
             ),
+            "github_workflow_directory": tmp_path / ".github" / "workflows",
             "github_workflow_patcher_yaml": None,
             "github_template_dict": {
                 "dependency_manager_version": "2.3.0",