## 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
```bash
pip install pypiper
```
### Quick Example
```python
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`
```python
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.
:::note[Example]
pm = PipelineManager(name="my_pipeline", outfolder="output/")
pm.run("samtools sort in.bam > out.bam", target="out.bam")
pm.report_result("read_count", 1000000)
pm.complete()
:::
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`
```python
@property
def halted -> bool
```
Whether the managed pipeline is in a paused/halted state.
##### `pipestat`
```python
@property
def pipestat -> PipestatManager
```
Access the PipestatManager for reporting results and managing status.
:::note[Example]
pm.pipestat.report(values={"reads": 1000}, record_identifier="sample1")
status = pm.pipestat.get_status("sample1")
:::
**Returns:**
- *PipestatManager*: Configured PipestatManager instance.
**Raises:**
- *PipestatError*: If pipestat was not initialized (no schema provided).
#### Methods
##### `__enter__`
```python
def __enter__()
```
Support use as a context manager.
:::note[Example]
with PipelineManager("test", "output/") as pm:
pm.run("echo hello", target="output/hello.txt")
# stop_pipeline() called automatically on clean exit
# fail_pipeline() called automatically on exception
:::
##### `__exit__`
```python
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`
```python
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.
:::note[Example]
retcode, maxmem = pm.callprint("samtools sort in.bam")
:::
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`
```python
def checkprint(cmd: str, shell: bool | None = None, nofail: bool = False) -> str
```
Run a command and return its stdout as a string.
:::note[Example]
genome_size = pm.checkprint("wc -l genome.fa")
version = pm.checkprint("samtools --version")
:::
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`
```python
def clean_add(regex: str | None, conditional: bool = False, manual: bool = False) -> None
```
Register files for automatic deletion when the pipeline succeeds.
:::note[Example]
pm.clean_add("intermediate.bam")
pm.clean_add("temp_*.txt", conditional=True)
pm.clean_add(None) # no-op, safe for unset variables
:::
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`
```python
def complete() -> None
```
Mark the pipeline as successfully completed and finalize.
:::note[Example]
pm.complete()
:::
Records elapsed time and success timestamp, runs cleanup of
intermediate files, and sets the pipeline status flag to 'completed'.
##### `critical`
```python
def critical(msg: str, *args: Any, **kwargs: Any) -> None
```
##### `debug`
```python
def debug(msg: str, *args: Any, **kwargs: Any) -> None
```
##### `error`
```python
def error(msg: str, *args: Any, **kwargs: Any) -> None
```
##### `fail_pipeline`
```python
def fail_pipeline(exc: Exception, dynamic_recover: bool = False) -> None
```
Stop the pipeline with a failure status and raise the given exception.
:::note[Example]
pm.fail_pipeline(Exception("BAM file is empty"))
:::
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`
```python
def fatal(msg: str, *args: Any, **kwargs: Any) -> None
```
##### `get_container`
```python
def get_container(image: str, mounts: str | list[str]) -> None
```
Start a Docker container for running pipeline commands.
:::note[Example]
pm.get_container("nsheff/refgenie", ["/data", "/ref"])
:::
**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`
```python
def get_elapsed_time() -> float
```
Calculate total pipeline runtime from the profile file.
:::note[Example]
total_seconds = pm.get_elapsed_time()
:::
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`
```python
def get_stat(key: str) -> Any
```
Retrieve a previously reported stat, loading from disk if needed.
:::note[Example]
trimmed = pm.get_stat("trimmed_reads")
rate = pm.get_stat("aligned_reads") / trimmed
:::
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`
```python
def halt(checkpoint: str | None = None, finished: bool = False, raise_error: bool = True) -> None
```
Pause the pipeline before its natural completion point.
:::note[Example]
pm.halt(checkpoint="alignment", finished=True)
:::
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`
```python
def info(msg: str, *args: Any, **kwargs: Any) -> None
```
##### `make_sure_path_exists`
```python
@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`
```python
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`
```python
def remove_container(container: str) -> None
```
Remove a Docker container by ID.
**Parameters:**
- `container` (*str*): Docker container ID string.
##### `report_object`
```python
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.
:::note[use pm.pipestat.report() directly instead:]
pm.pipestat.report(values={"peak_plot": {"path": "peaks.png"}})
:::
**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`
```python
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.
:::note[Example]
pm.report_result("aligned_reads", 1500000)
pm.report_result("alignment_rate", 0.95, nolog=True)
:::
**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`
```python
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.
:::note[Example]
pm.run("sort input.txt > output.txt", target="output.txt")
pm.run(["cmd1", "cmd2"], target="final.out")
pm.run("samtools index in.bam", target="in.bam.bai", clean=True)
:::
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`
```python
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`
```python
def stop_pipeline(status: str = COMPLETE_FLAG) -> None
```
Terminate the pipeline, recording time/memory stats and running cleanup.
:::note[Example]
pm.stop_pipeline() # defaults to 'completed' status
:::
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`
```python
@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`
```python
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.
:::note[Example]
pm.timestamp("### Alignment")
pm.timestamp("Reads trimmed", checkpoint="trim", finished=True)
:::
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`
```python
def warning(msg: str, *args: Any, **kwargs: Any) -> None
```