Report progress, intermediate output, and worker logs

Use an injected task context when worker code needs to report activity before returning its final result.

The submitter chooses the keyword name explicitly with context_keyword. Wetlands never changes a function call by guessing from its signature.

Complete example

from __future__ import annotations

import logging
from pathlib import Path

from wetlands import EnvironmentManager, EnvironmentSpec, ExecutionEvent, ExecutionEventKind, LocalPackage


def main(root: Path = Path("wetlands")) -> dict[str, object]:
    """Run a task that reports progress, output, and a worker log."""
    example_directory = Path(__file__).parent
    spec = EnvironmentSpec(
        python="3.12.*",
        conda=("pip",),
        local=(LocalPackage(example_directory),),
    )
    progress: list[tuple[int, int]] = []

    with EnvironmentManager(root=root) as manager:
        environment = manager.provision("docs-examples", spec).wait_for()
        with environment.start() as pool:
            task = pool.submit_import(
                "example_module:process_items",
                args=([1, 2, 3, 4],),
                context_keyword="task",
            )

            def report(event: ExecutionEvent) -> None:
                if event.kind is not ExecutionEventKind.UPDATE:
                    return
                if event.current is None or event.maximum is None or event.progress is None:
                    return
                point = (event.current, event.maximum)
                if progress and progress[-1] == point:
                    return
                progress.append(point)
                print(f"Progress: {event.current}/{event.maximum} ({event.progress:.0%})")

            task.listen(report)
            result = task.wait_for()
            outputs = task.outputs

    print(f"Intermediate output: {outputs['items_processed']} items")
    print(f"Result: {result}")
    return {"progress": progress, "outputs": outputs, "result": result}


if __name__ == "__main__":
    logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
    main()

The worker function used by the example calls task.update(), task.set_output(), and task.log(). Its source is in examples/example_module.py.

Representative output is:

Progress: 1/4 (25%)
Progress: 2/4 (50%)
Progress: 3/4 (75%)
Progress: 4/4 (100%)
INFO: Worker finished processing items
Intermediate output: 4 items
Result: [2, 4, 6, 8]

Listen only for progress updates

Every task also emits started, completion, failure, and cancellation events. Filter for ExecutionEventKind.UPDATE when a callback only renders progress. An intermediate-output change is also an update and repeats the latest numeric progress, so the example renders only changed current and maximum pairs.

Listeners may run on Wetlands background threads. A desktop application must forward UI changes through its toolkit's thread-safe mechanism.

Choose the right output channel

• Use update() for a human-readable status and numeric progress.
• Use set_output() for a small named value the caller may inspect through task.outputs.
• Use log() for diagnostic messages handled by the application's wetlands logger configuration.
• Return large values and NumPy arrays as the final task result.

Intermediate values are intentionally limited to small supported Python values. NumPy arrays are not supported as intermediate outputs.

See Injected task context for exact method signatures and validation rules.