Creating Custom Steps

Creating Custom Steps#

Learn how to create custom pipeline steps for your specific build, test, or deployment needs.

Step Basics#

Every step inherits from PipelineStep[ExecutionContext] and implements:

  • run() – Execute the step’s logic

  • update_execution_context() – Share state with subsequent steps

Minimal Example#

Create steps/hello_step.py:

from pypeline.domain.execution_context import ExecutionContext
from pypeline.domain.pipeline import PipelineStep


class HelloStep(PipelineStep[ExecutionContext]):
  def run(self) -> None:
    print("Hello from my custom step!")

  def update_execution_context(self) -> None:
    pass  # No state to share

Reference it in pypeline.yaml:

pipeline:
  - step: HelloStep
    file: steps/hello_step.py

Using Configuration#

Steps can receive configuration from the YAML:

from pypeline.domain.execution_context import ExecutionContext
from pypeline.domain.pipeline import PipelineStep


class ConfigurableStep(PipelineStep[ExecutionContext]):
  def run(self) -> None:
    config = self.config or {}
    message = config.get("message", "default")
    count = config.get("count", 1)
    for _ in range(count):
      print(message)

  def update_execution_context(self) -> None:
    pass

pipeline:
  - step: ConfigurableStep
    file: steps/my_step.py
    config:
      message: "Build started!"
      count: 3

Sharing State Between Steps#

Use ExecutionContext to pass data downstream:

from pathlib import Path

from pypeline.domain.execution_context import ExecutionContext
from pypeline.domain.pipeline import PipelineStep


class SetupStep(PipelineStep[ExecutionContext]):
  @property
  def tool_dir(self) -> Path:
    return self.output_dir / "tools"

  def run(self) -> None:
    # Install tools to a directory
    self.tool_dir.mkdir(exist_ok=True)

  def update_execution_context(self) -> None:
    # Add tool directory to PATH for subsequent steps
    # Called even when step is skipped, so compute path deterministically
    self.execution_context.add_install_dirs([self.tool_dir])

Access shared data in later steps:

class BuildStep(PipelineStep[ExecutionContext]):
  def run(self) -> None:
    # Tools from SetupStep are now in PATH
    executor = self.execution_context.create_process_executor(
      ["my-tool", "--version"]
    )
    executor.execute()

  def update_execution_context(self) -> None:
    pass

Dependency Management#

Implement get_inputs() and get_outputs() for smart rebuilds:

from pathlib import Path

from pypeline.domain.execution_context import ExecutionContext
from pypeline.domain.pipeline import PipelineStep


class CompileStep(PipelineStep[ExecutionContext]):
  def run(self) -> None:
    # Compile source files
    pass

  def get_inputs(self) -> list[Path]:
    return list(self.project_root_dir.glob("src/**/*.c"))

  def get_outputs(self) -> list[Path]:
    return [self.output_dir / "output.bin"]

  def update_execution_context(self) -> None:
    pass

The step only runs when inputs are newer than outputs.

Next Steps#

Contents