Feat/add custom command

docs/reference/cli.md

@@ -35,6 +35,8 @@ pypeline run [OPTIONS]
 | `--force-run` | FLAG | `false` | Force execution ignoring dependencies |
 | `--dry-run` | FLAG | `false` | Show what would run |
 | `-i`, `--input` | TEXT | — | Input as `key=value` (repeatable) |
+| `--command` | TEXT | — | Command appended as last step; pypeline waits for it |
+| `--application` | TEXT | — | Application started detached once the pipeline finished |
 
 ### `pypeline --version`
 
@@ -62,3 +64,21 @@ pypeline run -i env=prod -i debug=true
 # Preview without running
 pypeline run --print
 ```
+
+## Running Commands and Applications
+
+Both options use the environment collected by the pipeline: every `install_dirs` entry is in `PATH` and every `env_vars` entry is set. This is what makes them different from running the command yourself afterwards.
+
+`--command` is appended to the schedule as a regular step. It runs last, pypeline waits for it and fails if it fails. Because it is a step, it is skipped by `--dry-run` and supports `${{ inputs.<name> }}` placeholders for the inputs declared in the configuration file.
+
+```shell
+pypeline run --step CreateVEnv --command "pytest -v"
+pypeline run --command "ruff check ${{ inputs.target }}" -i target=src
+```
+
+`--application` adds no step. The pipeline runs, the application is started as a detached process and pypeline exits without waiting for it. Use it to launch editors, GUIs or shells that need the installed tools.
+
+```shell
+pypeline run --step ScoopInstall --application "code ."
+pypeline run --application "cmd"
+```

src/pypeline/domain/execution_context.py

@@ -30,13 +30,16 @@ def add_install_dirs(self, install_dirs: List[Path]) -> None:
     def add_env_vars(self, env_vars: Dict[str, Any]) -> None:
         self.env_vars.update(env_vars)
 
-    def create_process_executor(self, command: List[str | Path], cwd: Optional[Path] = None) -> SubprocessExecutor:
+    def create_process_env(self) -> Dict[str, str]:
         env = os.environ.copy()
         env.update(self.env_vars)
         env["PATH"] = os.pathsep.join([path.absolute().as_posix() for path in self.install_dirs] + [env["PATH"]])
+        return env
+
+    def create_process_executor(self, command: List[str | Path], cwd: Optional[Path] = None) -> SubprocessExecutor:
         # When started from a windows shell (e.g. cmd on Jenkins) the shell parameter must be set to True
         shell = True if os.name == "nt" else False
-        return SubprocessExecutor(command, cwd=cwd, env=env, shell=shell)
+        return SubprocessExecutor(command, cwd=cwd, env=self.create_process_env(), shell=shell)
 
     def create_artifacts_locator(self) -> ProjectArtifactsLocator:
         return ProjectArtifactsLocator(self.project_root_dir)

src/pypeline/main.py

@@ -1,18 +1,21 @@
+import os
+import shlex
+import subprocess
 import sys
 from pathlib import Path
-from typing import List, Optional
+from typing import Dict, List, Optional
 
 import typer
 from py_app_dev.core.exceptions import UserNotificationException
 from py_app_dev.core.logging import logger, setup_logger, time_it
 
 from pypeline import __version__
 from pypeline.domain.execution_context import ExecutionContext
-from pypeline.domain.pipeline import PipelineConfigIterator
+from pypeline.domain.pipeline import PipelineConfigIterator, PipelineStepConfig, PipelineStepReference
 from pypeline.domain.project_slurper import ProjectSlurper
 from pypeline.inputs_parser import InputsParser
 from pypeline.kickstart.create import KickstartProject
-from pypeline.pypeline import PipelineScheduler, PipelineStepsExecutor
+from pypeline.pypeline import PipelineScheduler, PipelineStepsExecutor, RunCommandClassFactory
 
 package_name = "pypeline"
 
@@ -21,6 +24,24 @@ def package_version_file() -> Path:
     return Path(__file__).parent / "__init__.py"
 
 
+def start_detached_process(command: List[str], env: Dict[str, str], cwd: Path) -> None:
+    """Start a process which outlives pypeline, e.g. an IDE which needs the tools installed by the pipeline in its PATH."""
+    logger.info(f"Starting detached process: {' '.join(command)}")
+    # A new process group (Windows) or session (POSIX) detaches the child, so it survives pypeline exiting.
+    if sys.platform == "win32":
+        creation_flags = subprocess.CREATE_NEW_PROCESS_GROUP
+    else:
+        creation_flags = 0
+    subprocess.Popen(
+        command,
+        cwd=cwd,
+        env=env,
+        shell=True if os.name == "nt" else False,  # noqa: S603 # The command comes from the user, same as for any other pipeline step.
+        creationflags=creation_flags,
+        start_new_session=os.name != "nt",
+    )
+
+
 app = typer.Typer(
     name=package_name, help="Configure and execute pipelines with Python (similar to GitHub workflows or Jenkins pipelines).", no_args_is_help=True, add_completion=False
 )
@@ -60,6 +81,8 @@ def run(
         "-i",
         help="Provide input parameters as key=value pairs (e.g., -i name=value -i flag=true).",
     ),
+    command: Optional[str] = typer.Option(None, help="Command to run as last pipeline step. Pypeline waits for it to finish."),
+    application: Optional[str] = typer.Option(None, help="Application to start once the pipeline finished. It is started as detached process, pypeline does not wait for it."),
 ) -> None:
     project_dir = project_dir.absolute()
     project_slurper = ProjectSlurper(project_dir, config_file)
@@ -75,7 +98,10 @@ def run(
         raise UserNotificationException("No pipeline found in the configuration.")
     # Schedule the steps to run
     steps_references = PipelineScheduler[ExecutionContext](project_slurper.pipeline, project_dir).get_steps_to_run(step, single)
-    if not steps_references:
+    if command:
+        command_step_class = RunCommandClassFactory().create_step_class(PipelineStepConfig(step="Command", run=command), project_dir)
+        steps_references.append(PipelineStepReference(None, command_step_class))
+    if not steps_references and not application:
         logger.info("No steps to run.")
         return
     # Parse the inputs
@@ -86,7 +112,14 @@ def run(
         inputs_dict = InputsParser.from_inputs_definitions(input_definitions).parse_inputs(inputs)
     else:
         inputs_dict = {}
-    PipelineStepsExecutor[ExecutionContext](ExecutionContext(project_dir, inputs=inputs_dict), steps_references, force_run, dry_run).run()
+    execution_context = ExecutionContext(project_dir, inputs=inputs_dict)
+    PipelineStepsExecutor[ExecutionContext](execution_context, steps_references, force_run, dry_run).run()
+    if application and not dry_run:
+        try:
+            application_command = shlex.split(application, posix=(os.name != "nt"))
+        except ValueError as exc:
+            raise UserNotificationException(f"Could not parse the application command '{application}': {exc}") from exc
+        start_detached_process(application_command, execution_context.create_process_env(), project_dir)
 
 def main() -> None:
     try:

tests/conftest.py

@@ -71,6 +71,9 @@ def run(self) -> None:
                     data = self.execution_context.data_registry.find_data(MyData)
                     if not data:
                         raise ValueError("Data not found")
+            class MyEnvStep(BaseStep):
+                def update_execution_context(self) -> None:
+                    self.execution_context.add_env_vars({"PYPELINE_TEST_VAR": "from_step"})
             class MyInputsChecker(BaseStep):
                 def run(self) -> None:
                     input = self.execution_context.inputs.get("my_input")