Package pypiper Documentation

Package Overview

The pypiper package provides a framework for building robust, restartable bioinformatics pipelines. It handles common pipeline tasks like checkpointing, logging, and resource monitoring.

Key Features

Automatic Checkpointing: Resume pipelines from where they left off
Resource Monitoring: Track memory and CPU usage
Result Reporting: Integrate with pipestat for standardized results
Container Support: Run commands in Docker containers
Pipeline Management: Built-in logging and status tracking

Installation

pip install pypiper

Quick Example

from pypiper import PipelineManager

# Initialize a pipeline
pm = PipelineManager(
    name="my_pipeline",
    outfolder="results/"
)

# Run a command
pm.run("echo 'Hello, world!'")

# Stop the pipeline
pm.stop_pipeline()

API Reference

PipelineManager Class

The main class for building and managing pipelines:

`PipelineManager`

class PipelineManager(name: str, outfolder: str, version: str | None = None, args: Any = None, multi: bool = False, dirty: bool = False, recover: bool = False, new_start: bool = False, force_follow: bool = False, cores: int = 1, mem: str = '1000M', config_file: str | None = None, output_parent: str | None = None, overwrite_checkpoints: bool = False, logger_kwargs: dict[str, Any] | None = None, pipestat_record_identifier: str | None = None, pipestat_schema: str | None = None, pipestat_results_file: str | None = None, pipestat_config: str | None = None, pipestat_pipeline_type: str | None = None, pipestat_validate_results: bool | None = None, pipestat_additional_properties: bool | None = None, pipestat_result_formatter: Callable | None = None, **kwargs: Any)

Manage a pipeline: run commands, track files, and report results.

PipelineManager handles command execution with file locking, target-based skipping for restartability, intermediate file cleanup, and result reporting via pipestat. Each command can declare a target file; if the target exists, the command is skipped. Lock files prevent parallel pipelines from colliding on the same target. Checkpoints (via timestamp()) enable start/stop control for partial reruns.

Parameters:

name (str): Pipeline name, used for output files, flags, and logging.
outfolder (str): Directory for pipeline results.
version (str | None): Pipeline version string. (default: None)
args (Any): Parsed argparse.Namespace; pypiper records these and extracts relevant options (recover, new_start, etc.). (default: None)
multi (bool): Allow multiple pipelines to share a pipestat results file. (default: False)
dirty (bool): Never auto-delete intermediate files. (default: False)
recover (bool): Overwrite lock files to restart a failed pipeline. (default: False)
new_start (bool): Rerun every command even if output exists. (default: False)
force_follow (bool): Always run follow functions even if command was skipped. (default: False)
cores (int): Number of processors. Default: 1. (default: 1)
mem (str): Memory limit with unit suffix [K|M|G|T]. Default: "1000M". (default: '1000M')
config_file (str | None): Path to pipeline YAML configuration file. (default: None)
output_parent (str | None): Parent directory for the output folder. (default: None)
overwrite_checkpoints (bool): Ignore checkpoint files (used by Pipeline class). (default: False)
logger_kwargs (dict[str, Any] | None): Keyword arguments for logmuse logger setup. (default: None)
pipestat_record_identifier (str | None): Record ID for pipestat reporting. (default: None)
pipestat_schema (str | None): Path to pipestat output schema. (default: None)
pipestat_results_file (str | None): Path to YAML file backend for results. (default: None)
pipestat_config (str | None): Path to pipestat configuration file. (default: None)
pipestat_pipeline_type (str | None): "sample" or "project". (default: None)
pipestat_validate_results (bool | None): Validate results against schema. None (default) auto-detects based on schema presence. (default: None)
pipestat_additional_properties (bool | None): Allow results not in schema. None (default) uses schema's own setting. (default: None)
pipestat_result_formatter (Callable | None): Callable for formatting reported results. (default: None)

Attributes

clean_initialized
cleanup_file
cleanup_list
cleanup_list_conditional
completed_procs
config
container
cores
cores1of2
cores1of2a
cores1of4
cores1of8
cores3of4
cores7of8
curr_checkpoint
dirty
force_follow
halt_on_next
javamem
last_timestamp
locks
mem
name
new_start
outfolder
output_parent
overwrite_checkpoints
overwrite_locks
peak_memory
pipeline_commands_file
pipeline_log_file
pipeline_profile_file
pipeline_stats_file
pipestat_pipeline_type
pipestat_record_identifier
pipestat_result_formatter
pipestat_results_file
pl_version
prev_checkpoint
proc_count
running_procs
starttime
stats_dict
status
testmode
wait

Properties

`halted`

@property
def halted -> bool

Whether the managed pipeline is in a paused/halted state.

`pipestat`

@property
def pipestat -> PipestatManager

Access the PipestatManager for reporting results and managing status.

Returns:

PipestatManager: Configured PipestatManager instance.

Raises:

PipestatError: If pipestat was not initialized (no schema provided).

Methods

`enter`

def __enter__()

Support use as a context manager.

`exit`

def __exit__(exc_type, exc_val, exc_tb)

Clean up pipeline on context manager exit.

Calls stop_pipeline() on clean exit, fail_pipeline() on exception.

`callprint`

def callprint(cmd: str, shell: bool | None = None, lock_file: str | None = None, nofail: bool = False, container: str | None = None) -> tuple[list[int | None], list[float]]

Execute a command, print it, and track memory usage.

Piped commands (|) are split into subprocesses so each can be memory- profiled independently. Shell mode (used for redirects > or wildcards *) runs the whole command in a shell, which prevents per-process memory tracking. Prefer shell=False (default) when possible for better profiling.

This is the low-level execution method; most users should use run() instead, which adds target-based skipping and file locking on top.

Parameters:

cmd (str): Shell command string.
shell (bool | None): Force shell mode. None auto-detects from pipes/redirects. (default: None)
lock_file (str | None): Lock file name for this execution. (default: None)
nofail (bool): If True, pipeline continues past nonzero return codes. (default: False)
container (str | None): Docker container name for execution. (default: None)

Returns:

tuple[list[int | None], list[float]]: Tuple of (return_codes_per_process, peak_memory_GB_per_process).

`checkprint`

def checkprint(cmd: str, shell: bool | None = None, nofail: bool = False) -> str

Run a command and return its stdout as a string.

Like callprint, but captures and returns stdout (uses subprocess.check_output instead of subprocess.call). Use this when you need a command's output as a Python variable.

Parameters:

cmd (str): Shell command string.
shell (bool | None): Force shell mode. None auto-detects based on pipes/redirects. (default: None)
nofail (bool): If True, pipeline continues past failure instead of halting. (default: False)

Returns:

str: Stripped stdout string from the command. Empty string in test mode.

`clean_add`

def clean_add(regex: str | None, conditional: bool = False, manual: bool = False) -> None

Passing None is a no-op, allowing safe calls on variables that may not have been assigned in conditional branches.

Parameters:

regex (str | None): Unix glob pattern or filename to delete. None is a no-op.
conditional (bool): Only delete if no other pipelines are running (checks for absence of flag files from other pipelines). (default: False)
manual (bool): Only add to the manual cleanup script, never auto-delete. Note: if the PipelineManager was created with dirty=True, all files are forced to manual cleanup regardless. (default: False)

`complete`

def complete() -> None

Mark the pipeline as successfully completed and finalize.

Records elapsed time and success timestamp, runs cleanup of intermediate files, and sets the pipeline status flag to 'completed'.

`critical`

def critical(msg: str, *args: Any, **kwargs: Any) -> None

`debug`

def debug(msg: str, *args: Any, **kwargs: Any) -> None

`error`

def error(msg: str, *args: Any, **kwargs: Any) -> None

`fail_pipeline`

def fail_pipeline(exc: Exception, dynamic_recover: bool = False) -> None

Stop the pipeline with a failure status and raise the given exception.

Terminates running subprocesses, writes cleanup script (but does not delete intermediate files), and sets status to 'failed'.

Parameters:

exc (Exception): Exception to raise after cleanup.
dynamic_recover (bool): Create recovery flag files alongside each active lock file. These flags signal a waiting pipeline to proceed rather than waiting indefinitely for a lock that will never be released. Used for job termination (e.g. SIGTERM from a cluster scheduler) rather than code errors. (default: False)

`fatal`

def fatal(msg: str, *args: Any, **kwargs: Any) -> None

`get_container`

def get_container(image: str, mounts: str | list[str]) -> None

Start a Docker container for running pipeline commands.

Parameters:

image (str): Docker image name (e.g. "nsheff/refgenie").
mounts (str | list[str]): Path or list of paths to mount into the container.

`get_elapsed_time`

def get_elapsed_time() -> float

Calculate total pipeline runtime from the profile file.

Parses the profile TSV to sum unique command runtimes (deduplicating reruns). Falls back to wall-clock estimate if profile is unavailable.

Returns:

float: Total runtime in seconds.

`get_stat`

def get_stat(key: str) -> Any

Retrieve a previously reported stat, loading from disk if needed.

Checks in-memory cache first, then reads from the stats YAML file. Returns None with a warning if the stat is not found.

Parameters:

key (str): Stat name to retrieve.

Returns:

Any: The stat value, or None if not found.

`halt`

def halt(checkpoint: str | None = None, finished: bool = False, raise_error: bool = True) -> None

Pause the pipeline before its natural completion point.

Sets status to 'paused' and raises PipelineHalt if raise_error is True.

Parameters:

checkpoint (str | None): Name of stage just reached or completed. (default: None)
finished (bool): Whether the indicated stage was just completed. (default: False)
raise_error (bool): Raise PipelineHalt to stop execution. (default: True)

`info`

def info(msg: str, *args: Any, **kwargs: Any) -> None

`make_sure_path_exists`

@staticmethod
def make_sure_path_exists(path: str) -> None

Create all directories in a path, no error if they already exist.

Parameters:

path (str): Directory path to create.

`process_counter`

def process_counter() -> int | str

Increment and return the process counter for logging.

Returns "Nf" (e.g. "3f") for commands inside a follow function, or the incremented integer count otherwise.

`remove_container`

def remove_container(container: str) -> None

Remove a Docker container by ID.

Parameters:

container (str): Docker container ID string.

`report_object`

def report_object(key: str, filename: str, anchor_text: str | None = None, anchor_image: str | None = None, annotation: str | None = None, nolog: bool = False, result_formatter: Callable | None = None, force_overwrite: bool = True) -> None

Report a file/image result via pipestat.

Parameters:

key (str): Result name.
filename (str): Path to the file (relative to output dir).
anchor_text (str | None): Link text or caption. Defaults to key. (default: None)
anchor_image (str | None): Path to thumbnail image (.png/.jpg). (default: None)
annotation (str | None): Annotation string. Defaults to pipeline name. (default: None)
nolog (bool): Suppress logging. (default: False)
result_formatter (Callable | None): Custom formatter callable. (default: None)
force_overwrite (bool): Overwrite existing results. (default: True)

Returns:

None: Formatted result string(s) from pipestat.

`report_result`

def report_result(key: str, value: Any, nolog: bool = False, result_formatter: Callable | None = None, force_overwrite: bool = True) -> Any

Report a key-value result to the pipeline stats file via pipestat.

Parameters:

key (str): Result name (must match schema if schema validation is on).
value (Any): Result value (str, int, float, dict, etc.).
nolog (bool): If True, suppress logging the result to the pipeline log. (default: False)
result_formatter (Callable | None): Custom formatter callable. Default: markdown. (default: None)
force_overwrite (bool): Overwrite existing results. Default: True. (default: True)

Returns:

Any: Formatted result string(s) from pipestat.

`run`

def run(cmd: str | list[str], target: str | list[str] | None = None, lock_name: str | list[str] | None = None, shell: bool | None = None, nofail: bool = False, clean: bool = False, follow: Callable | None = None, container: str | None = None, default_return_code: int | None = 0) -> int | None

Run a shell command with file-locking and target-based skipping.

If target exists, the command is skipped (restartability). If another process holds the lock, waits for it to finish. Creates lock files during execution to prevent parallel conflicts.

Follow functions run only when the command actually executes (target didn't exist), unless force_follow=True was set on the manager.

Parameters:

cmd (str | list[str]): Shell command string, or list of commands to run in sequence. A list of commands runs each sequentially, returning the max return code.
target (str | list[str] | None): Output file(s). If all exist, command is skipped. If None, lock_name is required. Multiple targets can be provided as a list; all must exist to skip. (default: None)
lock_name (str | list[str] | None): Explicit lock file name. Defaults to target-based name. Required if target is None (for commands with no output file). (default: None)
shell (bool | None): Force shell mode. None (default) auto-detects based on the presence of pipes (|) or redirects (>). (default: None)
nofail (bool): If True, pipeline continues past nonzero return codes. The failure is logged but does not halt the pipeline. (default: False)
clean (bool): If True, adds target to auto-cleanup list (deleted on pipeline success, kept with --dirty). (default: False)
follow (Callable | None): Callable to run after command execution. Skipped if the command was skipped (target existed), unless force_follow=True. (default: None)
container (str | None): Docker container name for execution. (default: None)
default_return_code (int | None): Return code when command is skipped (target existed). Default: 0. (default: 0)

Returns:

int | None: Return code (int). For command lists, the maximum return code.

`start_pipeline`

def start_pipeline(args: Any = None, multi: bool = False) -> None

Initialize pipeline logging, diagnostics, and status tracking.

Called automatically by init; rarely called directly.

Prints version/environment info, records git state, creates output folder, and sets pipeline status to 'running'. Logging is handled by a FileHandler (for pypiper messages) and per-command thread capture (for subprocess output) — no global fd manipulation needed.

`stop_pipeline`

def stop_pipeline(status: str = COMPLETE_FLAG) -> None

Terminate the pipeline, recording time/memory stats and running cleanup.

This is the underlying shutdown function. It reports Time and Success results, removes lock files, deletes intermediate files registered via clean_add(), and sets the status flag. Prefer complete() for success or fail_pipeline() for failure -- they call this internally.

Parameters:

status (str): Status flag string. Default: 'completed'. (default: COMPLETE_FLAG)

`time_elapsed`

@staticmethod
def time_elapsed(time_since: float) -> float

Return seconds elapsed since the given time (from time.time()).

Returns:

float: Elapsed seconds as float, rounded to nearest integer.

`timestamp`

def timestamp(message: str = '', checkpoint: str | None = None, finished: bool = False, raise_error: bool = True) -> None

Log a message with current time and elapsed time since last timestamp.

Messages starting with "###" are formatted as headings with surrounding newlines. If checkpoint is provided, creates a checkpoint file that enables start/stop control for pipeline reruns.

Checkpoint semantics: with finished=False (default), the checkpoint marks the start of a stage -- used with --stop-before to halt before a stage runs. With finished=True, it marks completion of a stage -- used with --stop-after to halt after a stage finishes. A typical pattern is to call timestamp() with just a checkpoint name before each stage (finished=False is implied), and pypiper handles the rest.

Parameters:

message (str): Message to log with the timestamp. (default: '')
checkpoint (str | None): Stage name for checkpoint file creation. (default: None)
finished (bool): True if the checkpoint stage just completed (retrospective); False (default) if the stage is just starting (prospective). (default: False)
raise_error (bool): Whether to raise PipelineHalt when a stop point is reached. (default: True)

`warning`

def warning(msg: str, *args: Any, **kwargs: Any) -> None