Metadata-Version: 2.4
Name: wofl
Version: 0.1.0
Summary: Composable tasks for Python workflows: deferred DAGs, caching, task-dirs, remote execution.
License-Expression: CC0-1.0
Project-URL: Homepage, https://github.com/ast-al/wofl
Project-URL: Source, https://github.com/ast-al/wofl
Project-URL: Issues, https://github.com/ast-al/wofl/issues
Keywords: workflow,pipeline,dag,caching,remote,hpc
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Operating System :: POSIX
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE.md
Provides-Extra: proctitle
Requires-Dist: setproctitle; extra == "proctitle"
Provides-Extra: format
Requires-Dist: black; extra == "format"
Provides-Extra: dev
Requires-Dist: pytest; extra == "dev"
Dynamic: license-file

# wofl:  composable tasks for Python workflows

wofl is a lightweight, Python-native alternative to heavyweight workflow engines
such as Nextflow or Snakemake.  The goal is to express entire pipelines in
regular Python while keeping caching, task isolation, retries, and remote
execution primitives close at hand.

## Table of Contents

- [Why use wofl?](#why-use-wofl)
- [Running shell pipelines safely from Python](#running-shell-pipelines-safely-from-python)
  - [Process substitution without a shell](#process-substitution-without-a-shell)
- [Network failures and transient errors](#network-failures-and-transient-errors)
- [Persistent caching/memoization](#persistent-cachingmemoization)
- [Parallel task execution with dependencies](#parallel-task-execution-with-dependencies)
  - [`@task:` a composition of `@deferred . @in_taskdir . @cached`.](#task-a-composition-of-deferred--in_taskdir--cached)
- [Scatter-gather workloads](#scatter-gather-workloads)
- [Running tasks on HPC clusters or remote servers](#running-tasks-on-hpc-clusters-or-remote-servers)
- [Running containerized workloads (Docker / Singularity / Apptainer / Conda)](docs/running_containerized_workloads.md)
- [Decorator application order](#decorator-application-order)
- [Full pipeline example](#full-pipeline-example)
- [Installation](#installation)

## Why use wofl?

- **Plain Python instead of a framework DSL.** `wofl` decorates functions rather
  than taking over your pipeline, so your debugger, type checker, tests, and IDE
  still see ordinary code. Adopt it one function at a time, with no migration
  project.
- **No business logic buried in bash strings.** Task bodies stay Python, while
  command-line tools still compose into UNIX pipelines. You keep shell power
  without making the shell your workflow language.
- **Safer by construction.** Common pipelines do not need a shell; when one is
  unavoidable, quoting and strict execution stay explicit.
- **Errors you can actually debug.** Failures surface as ordinary Python
  exceptions, locally or remotely, and remote jobs return the context needed to
  diagnose them. You debug your code, not an opaque workflow runtime.
- **Caching that understands your inputs.** Finished work survives reruns.
  Changes to task code or tracked file inputs invalidate affected results
  automatically.
- **Parallelism without a new mental model.** Build static and dynamic
  scatter-gather workflows from normal calls and return values instead of
  learning a channel grammar.
- **One decorator takes you remote.** Develop and debug a function locally, then
  run the same task body on a cluster or cloud job.
- **Zero required Python package dependencies.** Install or vendor one
  public-domain module built on the Python standard library wherever your
  pipeline runs.
- **A tiny API you can hold in your head.** A few composable decorators add
  caching, task isolation, retries, resource limits, or remote execution to
  existing functions, so you adopt only what you need.

## Running shell pipelines safely from Python
**Use-case:** pipe commands together (e.g., `cmd1 | cmd2`) but avoid subprocess with `shell=True`.

**Solution:** Use the `wofl.open_cmd()` context manager to compose command
pipelines.

Quote each value interpolated into a command string with `qw()` so that it
remains one command argument.

```python
import wofl
from wofl import qw

def eutils_fetch_uids(db: str, query: str) -> list[int]:
    """Pipe esearch output to efetch, returning UIDs."""
    with (
        wofl.open_cmd(f"esearch -db {qw(db)} -query {qw(query)}") as esearch,
        wofl.open_cmd("efetch -format uid", stdin=esearch.stdout) as efetch,
    ):
        return [int(uid) for uid in efetch.stdout if uid.strip()]
```

### Process substitution without a shell
Bash process-substitutions `<(cmd)` and `>(cmd)` can be expressed by referencing
the sub-processes `stdin` or `stdout` as `/dev/fd/{my_proc.stdout.fileno()}`.
Note: Unlike the silent process-substitution failures in bash, if a subprocess fails,
the `CalledProcessError` exception is raised.

```python
# Equivalent of: diff -u <(grep x file1) <(grep y file2)
with (
    wofl.open_cmd(f"grep x {qw(file1)}", ok_retcodes=(0, 1)) as sub1,
    wofl.open_cmd(f"grep y {qw(file2)}", ok_retcodes=(0, 1)) as sub2,
    wofl.open_cmd(
        f"diff -u /dev/fd/{sub1.stdout.fileno()} /dev/fd/{sub2.stdout.fileno()}",
        ok_retcodes=(0, 1),
    ) as diff,
):
    print(diff.stdout.read())

# Equivalent of: cmd | tee >(gzip > out.gz) | wc -l
with (
    open("out.gz", "wb") as out_gz,
    wofl.open_cmd("gzip", stdin=subprocess.PIPE, stdout=out_gz) as gz,
    wofl.open_cmd(cmd) as producer,
    wofl.open_cmd(
        ["tee", f"/dev/fd/{gz.stdin.fileno()}"],
        stdin=producer.stdout,
    ) as tee,
    wofl.open_cmd("wc -l", stdin=tee.stdout) as wc,
):
    print(wc.stdout.read())
```

**Strict-bash mode**

If you ever do need a shell for something `open_cmd` can't express, pass
`shell=True, executable="strict_bash"` to `open_cmd` or `from_cmd`
to execute with the following bash header prepended.
```bash
set -Eeu -o pipefail -o noclobber             # strict errors: ERR trap, exit on error, unset vars, pipe failures, no-overwrite with '>'
shopt -s inherit_errexit 2>/dev/null || true  # propagate -e into subshells and pipelines
export LC_ALL=C LANG=C                        # deterministic locale
IFS=$'\n\t'                                   # safe field splitting
```

```python
import wofl
from wofl import qw

def eutils_fetch_uids(db: str, query: str) -> list[int]:
    out = wofl.from_cmd(
        f"esearch -db {qw(db)} -query {qw(query)} | efetch -format uid",
        shell=True,
        executable="strict_bash",
    )
    return [int(uid) for uid in out.splitlines() if uid.strip()]
```
---

## Network failures and transient errors
**Use-case:** function non-deterministically fails due to network issues, API rate limits, or temporary service unavailability.

**Solution:** Use `@retry_on_exception` decorator.
```python
import asyncio

from wofl import retry_on_exception

@retry_on_exception(
    delay=lambda attempt: 2 ** attempt if attempt < 5 else None,
    condition=lambda e: isinstance(e, (ConnectionError, TimeoutError))
)
def fetch_remote_data(url: str) -> str:
    return requests.get(url, timeout=5).text

data = asyncio.run(fetch_remote_data("https://example.com/data"))
```

The retry policy is governed entirely by two arguments:

- `delay: (attempt) -> seconds`: sleep before attempt `attempt` (1 = first
  retry). Return `None`/`False` to stop retrying; `0` means no delay.
  **Default: retry exactly once, after 5 seconds.**
- `condition: (exception) -> bool`: which exceptions are retryable (default:
  all). `KeyboardInterrupt`, `MemoryError`, `SystemExit`, `GeneratorExit`,
  `RecursionError` are never retried; `OSError` subclasses (`ConnectionError`,
  `TimeoutError`) are retried when `condition` permits.

---

## Persistent caching/memoization
**Use-case:** `@functools.cache`-like functionality, but for the cache to be persistent across script re-runs.

**Solution:** Use `@in_taskdir` and `@cached` to cache results with automatic invalidation:
```python
import asyncio

from wofl import cached, in_taskdir
from pathlib import Path

@in_taskdir(work_dir="./work")
@cached
def uppercase_file(input_file: Path) -> Path:
    output = Path("result.txt")
    output.write_text(input_file.read_text().upper())
    return output

output = asyncio.run(uppercase_file(Path("input.txt")))
```
`@in_taskdir`
- Creates a directory `task_dir = {work_dir}/{module}/{function}/{args_hash}/` and `chdirs` to it for the duration of the function call.
- It also creates `./tmp` in the taskdir and deletes it afterward. Transient work files should go there.
- An inter-process lock prevents identical task runs from using the same task directory concurrently.
- Any relative `Path`s in the input args are rebased to be valid within the taskdir.
- Upon return, any relative `Path`s in the return-value are rebased relative to the original CWD.

`@cached` memoizes the function-call in `{task_dir}/.wofl/cache.pkl.zst`, containing
- function source and wofl version hash
- metadata and content-hashes of input files passed as `Path` in function's args
- invocation's start and stop timestamps
- serialized function's return value

The cache is automatically invalidated by:
- source code changes, detected via code digest
- a changed wofl version
- input file(s) modifications, detected via mtime and content hash (note: directories are not content-hashed)
- cache expiration, if specified with `@cached(expires_in=...)` (e.g. if the function access a remote file or makes a db query)

Globals, external tools, environments, directory contents, and remote data are
not tracked automatically. Pass values that represent such dependencies as
ordinary task arguments.

When a task runs as part of a `@deferred` DAG, `{task_dir}/.wofl/deps.txt`
additionally records the task-dirs of the tasks it directly depends on
(one task-dir path per line, relative to the work-dir root), for post-mortem
tracking of what fed into a given task-run.

---

## Parallel task execution with dependencies
**Use-case:** multiple tasks with dependencies that you want to execute in parallel, like with `make -j4`

**Solution:** Use `@deferred` to build and execute task DAGs in parallel:
```python
import asyncio
import wofl

@wofl.deferred
def t_process_chunk(data: str) -> str:
    return data.upper()

@wofl.deferred
def t_join_results(chunks: list[str]) -> str:
    return " | ".join(chunks)

async def main():
    # Build DAG: process two chunks in parallel, then join
    d_chunk1 = t_process_chunk("hello")
    d_chunk2 = t_process_chunk("world")
    d_joined = t_join_results([d_chunk1, d_chunk2])

    output = await wofl.compute(d_joined, max_workers=4)
    print(output)  # "HELLO | WORLD" (both chunks processed in parallel)

asyncio.run(main())
```

The `@deferred` function call does not execute the wrapped function,
but instead captures it and the args and returns it immediately
as `Deferred` object that can later be computed in a `ProcessPoolExecutor`
or in-place. The root `Deferred` object effectively holds
the execution plan for the DAG of dependent subtasks.

If any of the bound args are themselves Deferred objects,
(e.g. `d_joined` holds a list of two `Deferred`s as args)
they are recursively computed ahead, and are computed in parallel;
the dependent `Deferred` is computed when the dependencies complete,
and the `Deferred` args materialize as function's return types.

Note: as a convention, we name the task-functions that return `Deferred`s
with `t_` prefix (tasks), and name the deferred-taskrun objects themselves
with `d_` prefix (deferred-taskrun).

`wofl.compute(deferred_task, max_workers=...)` is an awaitable convenience
wrapper that creates the `ProcessPoolExecutor` for you, submits the task graph,
and returns the root result. It defaults to `os.cpu_count()` workers, and always
uses an explicit fork policy, which the `@throttled` machinery expects for
inheriting resource semaphores. To keep the executor for other work, manage it
yourself:
```python
import asyncio
import concurrent.futures as cf

with cf.ProcessPoolExecutor(max_workers=4) as executor:
    output = asyncio.run(d_joined.compute(executor))
```


---

### `@task:` a composition of `@deferred . @in_taskdir . @cached`.
Now you can forget the other three - they're just building blocks of `@task`.

```python
import asyncio

# @deferred
# @in_taskdir
# @cached
# ----------
@task
def t_getfasta(ids: list[int]) -> Path:
    prots_fa = Path("prots.fa")
    ids_txt = "\n".join(map(str, ids))
    subprocess.run(["getfasta", "-o", "prots.fa"], input=ids_txt.encode(), check=True)
    return prots_fa

path_to_fasta = asyncio.run(t_getfasta(ids).compute())  # in-place, no executor

```
This creates the output file in `{work_root}/{module}/{function}/{args_hash}/prots.fa`.
As mentioned above `@in_taskdir` context-manager rebases the return-path to be relative
to the original directory upon return.

---

## Scatter-gather workloads
**Use-case:** A large input that's more efficient to split into chunks, process in parallel, then combine.

**Solution:** Define a @task-decorated scatter-function that returns a list of tasks (dynamic DAG):
```python
import asyncio

import wofl
from wofl import qw

def run(cmd):
    subprocess.run(shlex.split(cmd), check=True)

@wofl.task  # aligns single batch, producing a .hits file.
def t_align_prots(fasta: Path, db: Path) -> Path:
    hits_path = Path(fasta.stem).with_suffix(".hits")
    run(
        f"time diamond blastp --query {qw(fasta)} --db {qw(db)} --out {qw(hits_path)} "
        "--fast --threads 32 --header simple --quiet "
        "--outfmt 6 qseqid qlen qcovhsp sseqid slen scovhsp pident"
    )
    return hits_path

@wofl.task  # splits input fasta into batches of specified size
def t_split_fasta(fasta: Path, batch_size=10000) -> list[Path]:
    run(f"seqkit split --force -s {qw(batch_size)} -O ./out {qw(fasta)}")
    return [
        Path(f"./out/{fa}")
        for fa in os.listdir("out")
        if str(fa).endswith(".fa")
    ]

@wofl.task  # returns list of subtasks, each processing a single batch.
def t_scatter_align_prots(batches: list[Path], db: Path) -> list[Path]:
    return [t_align_prots(batch, db) for batch in batches]

@wofl.task  # combines the results from all batches into one file.
def t_gather_results(paths: list[Path]) -> Path:
    combined_path = Path("combined.hits")
    combined_path.write_bytes(b"".join(path.read_bytes() for path in paths))
    return combined_path

# workflow
d_fasta_batches = t_split_fasta(fasta=fasta_path)
d_hits_paths    = t_scatter_align_prots(d_fasta_batches, db=db_path)
d_combined_hits = t_gather_results(d_hits_paths)

combined_hits_result : Path = asyncio.run(wofl.compute(d_combined_hits, max_workers=4))
```

---

## Running tasks on HPC clusters or remote servers
**Use-case:** orchestrate and dispatch long-running tasks to an HPC cluster (Slurm, SGE) or remote server, or on the cloud.

**Solution:** Use `@remote` to have the wrapped function execute on HPC schedulers or SSH:

```python
import asyncio

import wofl

run_on_sge_1gig = wofl.remote(
    submit_cmd="qrsh -m n -V -l mem_free=1G",
    python3="/opt/python-all/bin/python3"  # or e.g. "singularity exec ... /path/in/container/bin/python3"
)                                          # to execute with docker/singularity/apptainer, etc.

@wofl.task
@run_on_sge_1gig
def foo(i: int) -> str:
    import socket
    print(f"Hello from {socket.gethostname()}!")
    open("out_file.txt", "w").write("hello\n")
    return str(i)

assert asyncio.run(foo(42).compute()) == "42"
```

```sh
>>time python3 remote_example.py
Hello from sge12345!

real 0m3.377s
user 0m0.086s
sys  0m0.050s
```

The second execution returns quickly and does not print to stdout,
as it just returns the cached value.
```sh
>>time python3 remote_example.py
real 0m0.129s
user 0m0.071s
sys  0m0.019s
```

```
>>tree -a ./work
./work
`-- remote_example
    `-- foo
        `-- 9afe8dd7da54e50ae64c26798a5ed5d4551476ef
            |-- .wofl
            |   |-- cache.pkl.zst
            |   |-- cache.pkl.zst.sig
            |   |-- deps.txt
            |               `-- worker_node
            |       |-- host.info.json
            |       |-- host.meminfo.txt
            |       |-- host.rusage.json
            |       |-- host.top.txt
            |       |-- out.pkl.zst
            |       |-- stderr.zst
            |       |-- stdout.zst
            |       |-- inp.pkl.zst
            |       `-- task.py
            `-- out_file.txt
```

`@remote` renders the task - the function and its arguments - into a single
self-contained, human-readable `task.py` script, dispatches it to the remote
scheduler to execute the task, and pickles the return value (or the raised
exception, with a remote-host traceback) back into `out.pkl.zst`.

Dispatch is coordinated on the caller's event loop. SSH and blocking HPC
launchers are awaited asynchronously; detached HPC and cloud completion are
interval-polled, with blocking status and file operations run in threads. A
`@task @remote` deferred never occupies a ProcessPoolExecutor worker, so many
remote jobs can be in flight concurrently from a single process.

Task payloads must be self-contained: the function and same-file functions
it references are embedded, and referenced modules/imported functions are recreated.
Payload functions must use `def`; `async def` payloads are rejected before submission.
Other closures or globals from the submitting module are not shipped. Arguments
are Python-native values - scalars, `Path`s, and `tuple`/`list`/`dict`/
`set`/`frozenset` containers thereof - so that they render trivially as source.
Otherwise they are pickled to `inp.pkl.zst` next to `task.py` and
deserialized on the worker, so any picklable argument works (numpy arrays,
custom classes, etc.); heavy-weight inputs should still be passed as files.

The return value is unpickled on the submitting host with a restricted
unpickler, so results must be Python-native too (arguments' types, plus
`datetime`/`Decimal`/`UUID`/`os.stat_result` and built-in exceptions).
Custom classes - dataclasses, numpy arrays, enums - can't be returned;
convert them to native types (e.g. `tuple(x.tolist())`) at the end of the
remote function, or serialize to file(s) and return as `Path`.

To run on the cloud, `remote()` takes the path to the writable S3 or GCS bucket,
which is used as intermediate storage to exchange the input and output files between
the local and remote environments, which happens behind the scenes.

For containerized workloads on HPC, provide the containerized python3 interpreter.

Supported remote backends:
```
- SSH           : ssh user@host
- HPC schedulers: srun, sbatch, qrsh, qsub, bsub, flux, condor_submit
- AWS Batch     : aws batch submit-job --job-queue Q --job-definition D
- AWS ECS       : aws ecs run-task --cluster C --task-definition T
- GCP Batch     : gcloud batch jobs submit --location L
- GCP Cloud Run : gcloud run jobs execute JOB --region R
```

For Cloud Run, configure Bash as the deployed job's container command. Wofl
passes the remote script to that command as `-c <script>`.

---

## Decorator application order

The order of decorators is important.

```python
from pathlib import Path

from wofl import (
    cached,
    deferred,
    in_taskdir,
    remote,
    register_resource,
    retry_on_exception,
    throttled,
)

register_resource("banana", capacity=4)  # or e.g. "cpu"

# Parallel @throttled tasks require a fork-based process pool, which the
# @throttled machinery expects; wofl.compute() always uses it:
#     await wofl.compute(workflow, max_workers=4)

# @deferred
# @in_taskdir(work_dir="./work")
# @cached(expires_in=3600, clear_dir=True, file_checksum_fn="sha1")
# @task is a shorthand for the three decorators above.
@task(work_dir="./work", expires_in=None, clear_dir=False, file_checksum_fn="sha1")
@throttled(
    resource="banana"  # block until at least two bananas are available,
    n=2                # and atomically acquire for the duration of the
)                      # function execution.
@retry_on_exception(
    delay=lambda attempt: min(30, 2 ** attempt) if attempt < 5 else None,
    condition=lambda exc: isinstance(exc, (ConnectionError, TimeoutError)),
)
@remote(
    submit_cmd="qsub -m n -V -l mem_free=1G",
    python3=sys.executable,
)
def align_batch(fasta: Path, db: Path) -> Path:
    ...
```

`@cached`, `@in_taskdir`, `@remote`, `@retry_on_exception`, and `@throttled`
always create async functions. Await their calls, or use `asyncio.run()` at a
synchronous program boundary. `@deferred` and `@task` are the
exceptions: their calls synchronously construct DAG nodes, which execute when
you await `.compute()` or `wofl.compute()`.

---

[Full pipeline example](examples/align_prots.py)

---

## Installation

wofl has no required Python package dependencies. It requires Python 3.11 or
newer and a POSIX host.

Install the `zstd` command-line executable before you use the default
`@cached` or `@task` behavior. Staged SSH and cloud runs also require Bash and
tar on the submission and remote hosts. Run Python without `-O` and without
`PYTHONOPTIMIZE`; wofl uses assertions for runtime validation.

Install from PyPI:

```sh
pip install wofl
```

Or, if you use uv:
```sh
uv add wofl
```

Install directly from GitHub:

```sh
pip install git+https://github.com/ast-al/wofl.git
```

Or, if you use uv:
```sh
uv add git+https://github.com/ast-al/wofl.git
```

Or with the following shebang in your `pipeline.py`
```python
#!/usr/bin/env -S uv run --with git+https://github.com/ast-al/wofl.git
```

Note: for `@remote` runs, wofl only needs to be installed on the
submission host (where you launch the workflow). Remote hosts need Python 3.11
or newer and `zstd`; staged SSH and cloud hosts also need Bash and tar. wofl
ships its code as a self-contained `task.py` script.

Alternatively, since the library is a single module and public-domain licensed
(CC0-1.0), you can copy `src/wofl.py` directly into your project and vendor it,
if you prefer to avoid a dependency.
