Metadata-Version: 2.4
Name: wofl
Version: 0.2.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)
- [Execution graph recovery](#execution-graph-recovery)
- [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)
- [Throttled execution (resource limits)](#throttled-execution-resource-limits)
- [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: seq 1000 | tee >(gzip > out.gz) | wc -l
import subprocess

with (
    open("out.gz", "wb") as out_gz,
    wofl.open_cmd("gzip", stdin=subprocess.PIPE, stdout=out_gz) as gz,
    wofl.open_cmd("seq 1000") 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 (including the source of
  referenced top-level functions and the value of referenced picklable
  globals, transitively)
- 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)

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

## Execution graph recovery

When a task runs as part of a `@deferred` DAG, `{task_dir}/.wofl/deps.txt`
records the task-dirs of the tasks it directly depends on - one task-dir path
per line, relative to the work-dir root. Each task-dir is a DAG node and its
`deps.txt` holds its edges, so walking all `{work_dir}/**/.wofl/deps.txt`
files recovers the full graph, including transitive dependencies. Files are
written on the submission host when the DAG is scheduled (before tasks run),
so they reflect the planned execution order, and cache-served tasks are still
listed.

Recover the reachable execution graph from a root task directory with
`show-dag`:

```sh
python -m wofl show-dag ./work/<module>/<function>/<args_hash>
python -m wofl show-dag ./work/<module>/<function>/<args_hash> --out-fmt=tsv
```

The command follows `.wofl/deps.txt` manifests transitively. The default YAML
output has two root-level nodes:
Task paths omit the module prefix and use `<task>/<args_hash>`.

- `dag:` - the dependency forest. Each tree nests dependencies below their
  consuming task. If multiple tasks use one dependency, their nodes contain a
  reference and the command expands the dependency as another top-level tree.
  Terminal nodes use the `terminal` state, and deleted task directories use the
  `truncated` state.
- `tasks:` - every participating taskrun, keyed by its `<task>/<args_hash>`
  path, with the module prefix omitted and the `Path` values found in its
  arguments (`input-files:`). Each `Path` in its return value (`output-files:`)
  includes the file size and hash stored in the cache.

`--out-fmt=tsv` emits tab-separated `task`/`parent-task` rows instead:
`parent-task` is the task's direct dependency, terminal tasks use `none`, and
deleted task directories use `truncated`.

---

## 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 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)

# 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 = wofl.run(d_joined, max_workers=4)
print(output)  # "HELLO | WORLD" (both chunks processed in parallel)
```

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.run(deferred_task, max_workers=...)` executes the task graph and returns
the root result. It creates the `ProcessPoolExecutor` for you, defaults to
`os.cpu_count()` workers, and always uses an explicit fork policy, which the
`@throttled` machinery expects for inheriting resource semaphores.

`max_workers` limits the number of *local* tasks running concurrently; `@remote`
tasks run inline on the event loop (polling their backends) and do not consume
pool workers, so `max_workers` does not bound how many remote jobs are in
flight. To limit remote concurrency, use
[`@throttled`](#throttled-execution-resource-limits).

To keep the executor for other work, manage it yourself and await `.compute()`:
```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 subprocess
from pathlib import Path

import wofl
from wofl import task

# @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

ids = [1, 2, 3]
path_to_fasta = wofl.run(t_getfasta(ids))

```
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 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 = wofl.run(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 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 wofl.run(foo(42)) == "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
            |       |-- inp.pkl.zst
            |       |-- out.pkl.zst
            |       |-- stderr.zst
            |       |-- stdout.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`. The approach
is similar to Netxflow's `.command.run` and `.command.sh`. `task.py` runs
the packaged task and generates `out.pkl.zst` containing the task-function's
return-value or serialized raised exception with backtrace.
This is the equivalent of Nextflow's `.exitcode`. The exception is re-raised
locally with the full remote backtrace.

Nextflow decouples the cluster configuration from the pipeline definition.
Equivalently, the `@remote` parameters need not be hardcoded, and instead
come from some config-dictionary backed by some yaml-file, or however you desire.
```py
remote_smallhost = wofl.remote(
    submit_cmd=remote_config["sge"]["small"],
    python3=remote_config["apptainer-run-command"] + " python3 -B",
)

@wofl.task
@remote_smallhost
def my_task(i: int) -> str:
    ...
```

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>`.

---

## Throttled execution (resource limits)

`@throttled` bounds how many tasks consuming a shared resource run concurrently -
GPUs, memory, remote job slots, etc. Unlike `max_workers` (which only limits
local process-pool tasks), `@throttled` can also be used to limit `@remote`
tasks, which run on the event loop and never occupy pool workers.

Register each resource with its capacity, then attach `@throttled` to the tasks
that consume it:

```python
import wofl

wofl.register_resource("gpu", capacity=4)

@wofl.task
@wofl.throttled(resource="gpu", n=2)  # needs 2 GPUs; blocks until available
def train_model(config):
    ...
```

`n` units are acquired atomically for the duration of the call and released on
return, so at most `capacity // n` such tasks run at once. Register each resource
once, before submitting work - re-registering a name replaces its semaphore (with
a warning), which can diverge limits while throttled tasks already hold slots.

Because `@throttled` shares its semaphore with worker processes via fork,
throttled tasks require a fork-based process pool - `wofl.run()` always uses one.

---

## 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.run() always uses it:
#     wofl.run(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 call `wofl.run()` (or `await` `.compute()` with your own event loop/executor).

---

[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.

wofl logs cache hits/misses, task start/finish, and retry decisions through the
standard `logging` module but attaches no handler of its own, so those messages
are invisible until logging is configured. Call `wofl.enable_logging()` once to
see them on stderr (pass `file=` to write elsewhere), or configure the
`"wofl"` logger yourself:

```python
import wofl
wofl.enable_logging()  # optional; shows task/cache/retry progress on stderr
```

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.
