API Reference

Contents

API Reference#

Auto-generated documentation for pypeline’s public Python API.

Core Classes#

ExecutionContext#

class pypeline.domain.execution_context.ExecutionContext(project_root_dir: pathlib.Path, install_dirs: List[pathlib.Path] = <factory>, data_registry: py_app_dev.core.data_registry.DataRegistry = <factory>, inputs: Dict[str, Any] = <factory>, env_vars: Dict[str, Any] = <factory>)[source]#
add_env_vars(env_vars: Dict[str, Any]) → None[source]#
add_install_dirs(install_dirs: List[Path]) → None[source]#
create_artifacts_locator() → ProjectArtifactsLocator[source]#
create_process_executor(command: List[str | Path], cwd: Path | None = None) → SubprocessExecutor[source]#
data_registry: DataRegistry#
env_vars: Dict[str, Any]#
get_input(name: str) → Any | None[source]#
inputs: Dict[str, Any]#
install_dirs: List[Path]#
project_root_dir: Path#

PipelineStep#

class pypeline.domain.pipeline.PipelineStep(execution_context: TExecutionContext, group_name: str | None, config: Dict[str, Any] | None = None)[source]#

One can create subclasses of PipelineStep that specify the type of ExecutionContext they require.

get_needs_dependency_management() → bool[source]#

If false, the step executor will not check for outdated dependencies. This is useful for steps consisting of command lines which shall always run.

property output_dir: Path#
abstract update_execution_context() → None[source]#

Even if the step does not need to run ( because it is not outdated ), it can still update the execution context.

A typical use case is for steps installing software that need to provide the install directories in the execution context even if all tools are already installed.

PipelineLoader#

class pypeline.domain.pipeline.PipelineLoader(pipeline_config: List[PipelineStepConfig] | OrderedDict[str, List[PipelineStepConfig]], project_root_dir: Path, step_class_factory: StepClassFactory[TPipelineStep] | None = None)[source]#
load_steps_references() → List[PipelineStepReference[TPipelineStep]][source]#

PipelineStepConfig#

class pypeline.domain.pipeline.PipelineStepConfig(step: str, file: str | None = None, module: str | None = None, class_name: str | None = None, run: str | List[str] | NoneType = None, description: str | None = None, timeout_sec: int | None = None, config: Dict[str, Any] | None = None)[source]#
class_name: str | None = None#

Step class name

config: Dict[str, Any] | None = None#

Custom step configuration

description: str | None = None#

Step description

file: str | None = None#

Path to file with step class

classmethod from_dict(d, *, dialect=None)#
module: str | None = None#

Python module with step class

run: str | List[str] | None = None#

Command to run. For simple steps that don’t need a class.

Single command:

run: ruff check .

Or as list:

run: [pytest, -v, --cov]

Multiple commands (GitHub Actions style):

run: |
  ruff check .
  pytest -v --cov
step: str#

Step name or class name if file is not specified

timeout_sec: int | None = None#

Step timeout in seconds

to_dict()#

Orchestration#

PipelineScheduler#

class pypeline.pypeline.PipelineScheduler(pipeline: List[PipelineStepConfig] | OrderedDict[str, List[PipelineStepConfig]], project_root_dir: Path)[source]#

Schedules which steps must be executed based on the provided configuration.

  • If a step name is provided and the single flag is set, only that step will be executed.

  • If a step name is provided and the single flag is not set, all steps up to the provided step will be executed.

  • In case a command is provided, only the steps up to that command will be executed.

  • If no step name is provided, all steps will be executed.

static create_pipeline_loader(pipeline: List[PipelineStepConfig] | OrderedDict[str, List[PipelineStepConfig]], project_root_dir: Path) → PipelineLoader[PipelineStep[TExecutionContext]][source]#
static filter_steps(pipeline_config: List[PipelineStepConfig] | OrderedDict[str, List[PipelineStepConfig]], step_names: List[str] | None, single: bool) → List[PipelineStepConfig] | OrderedDict[str, List[PipelineStepConfig]][source]#

Filters the pipeline steps based on the provided step names.

If no step names are provided, all steps are returned. When single is True, only the named steps are returned, otherwise all steps up to the last named step are returned.

get_steps_to_run(step_names: List[str] | None = None, single: bool = False) → List[PipelineStepReference[PipelineStep[TExecutionContext]]][source]#

PipelineStepsExecutor#

class pypeline.pypeline.PipelineStepsExecutor(execution_context: TExecutionContext, steps_references: List[PipelineStepReference[PipelineStep[TExecutionContext]]], force_run: bool = False, dry_run: bool = False)[source]#

Executes a list of pipeline steps sequentially.

property artifacts_locator: ProjectArtifactsLocator#
run() → None[source]#

Built-in Steps#

CreateVEnv#

class pypeline.steps.create_venv.CreateVEnv(execution_context: ExecutionContext, group_name: str, config: Dict[str, Any] | None = None)[source]#
DEFAULT_PACKAGE_MANAGER = 'uv>=0.6'#
DEFAULT_PYTHON_EXECUTABLE = 'python311'#
SUPPORTED_PACKAGE_MANAGERS: ClassVar[Dict[str, List[str]]] = {'pipenv': ['Pipfile', 'Pipfile.lock'], 'poetry': ['pyproject.toml', 'poetry.lock'], 'uv': ['uv.lock', 'pyproject.toml']}#
property bootstrap_config_file: Path#
get_config() → dict[str, str] | None[source]#

Get runnable configuration.

(!) Do NOT put sensitive information in the configuration. It will be stored in a file.

get_inputs() → List[Path][source]#

Get runnable dependencies.

get_name() → str[source]#

Get runnable name.

get_needs_dependency_management() → bool[source]#

If false, the step executor will not check for outdated dependencies. This is useful for steps consisting of command lines which shall always run.

get_outputs() → List[Path][source]#

Get runnable outputs.

has_bootstrap_config() → bool[source]#

Check if user provided any bootstrap-specific configuration.

property install_dirs: List[Path]#
property package_manager_name: str#
property python_executable: str#

Get python executable to use.

Priority: 1. Input from execution context (execution_context.get_input(“python_version”)) 2. User-specified python_executable config 3. Auto-detect from python_version config 4. Current Python interpreter (sys.executable)

run() → int[source]#

Run and return exit code.

property target_internal_bootstrap_script: Path#
update_execution_context() → None[source]#

Even if the step does not need to run ( because it is not outdated ), it can still update the execution context.

A typical use case is for steps installing software that need to provide the install directories in the execution context even if all tools are already installed.

WestInstall#

class pypeline.steps.west_install.WestInstall(execution_context: TContext, group_name: str, config: dict[str, Any] | None = None)[source]#
get_config() → dict[str, str] | None[source]#

Get runnable configuration.

(!) Do NOT put sensitive information in the configuration. It will be stored in a file.

get_id() → str[source]#

Return unique identifier for dependency tracking (.deps.json filename).

get_inputs() → list[Path][source]#

Get runnable dependencies.

get_name() → str[source]#

Get runnable name.

get_outputs() → list[Path][source]#

Get runnable outputs.

property installed_dirs: list[Path]#
run() → int[source]#

Run and return exit code.

update_execution_context() → None[source]#

Even if the step does not need to run ( because it is not outdated ), it can still update the execution context.

A typical use case is for steps installing software that need to provide the install directories in the execution context even if all tools are already installed.

ScoopInstall#

class pypeline.steps.scoop_install.ScoopInstall(execution_context: TContext, group_name: str, config: dict[str, Any] | None = None)[source]#
get_inputs() → list[Path][source]#

Get runnable dependencies.

get_name() → str[source]#

Get runnable name.

get_outputs() → list[Path][source]#

Get runnable outputs.

property install_dirs: list[Path]#
run() → int[source]#

Run and return exit code.

update_execution_context() → None[source]#

Even if the step does not need to run ( because it is not outdated ), it can still update the execution context.

A typical use case is for steps installing software that need to provide the install directories in the execution context even if all tools are already installed.

GenerateEnvSetupScript#

class pypeline.steps.env_setup_script.GenerateEnvSetupScript(execution_context: ExecutionContext, group_name: str | None, config: Dict[str, Any] | None = None)[source]#
get_inputs() → List[Path][source]#

Get runnable dependencies.

get_name() → str[source]#

Get runnable name.

get_needs_dependency_management() → bool[source]#

If false, the step executor will not check for outdated dependencies. This is useful for steps consisting of command lines which shall always run.

get_outputs() → List[Path][source]#

Get runnable outputs.

run() → None[source]#

Run and return exit code.

update_execution_context() → None[source]#

Even if the step does not need to run ( because it is not outdated ), it can still update the execution context.

A typical use case is for steps installing software that need to provide the install directories in the execution context even if all tools are already installed.

Contents