Use Wetlands with asyncio

Provisioning operations and execution tasks can be awaited directly. They adapt completion to the event loop that is currently running.

Worker-pool startup and shutdown and manager shutdown are blocking lifecycle calls. Run those calls with asyncio.to_thread() so they do not block the event loop.

Complete example

from __future__ import annotations

import asyncio
from pathlib import Path

import numpy as np

from wetlands import EnvironmentManager, EnvironmentSpec, LocalPackage


async def main(root: Path = Path("wetlands")) -> None:
    example_directory = Path(__file__).parent
    manager = EnvironmentManager(root=root)
    try:
        operation = manager.provision(
            "async-numpy-example",
            EnvironmentSpec(
                python="3.12.*",
                conda=("numpy>=2", "pip"),
                local=(LocalPackage(example_directory),),
            ),
        )

        async def report() -> None:
            async for event in operation.events():
                print(event.message)

        reporter = asyncio.create_task(report())
        environment = await operation
        await reporter

        workers = await asyncio.to_thread(environment.start)
        try:
            image = np.arange(16, dtype=np.float32).reshape(4, 4)
            mask = await workers.submit_import(
                "example_module:threshold",
                kwargs={"image": image, "value": 7.5},
            )
            print(mask)
        finally:
            await asyncio.to_thread(workers.close)
    finally:
        await asyncio.to_thread(manager.close)


if __name__ == "__main__":
    asyncio.run(main())

Wetlands does not create, run, or stop the application's event loop.

Provisioning messages vary by installation, but the example ends by printing this mask:

[[False False False False]
 [False False False False]
 [ True  True  True  True]
 [ True  True  True  True]]

Consume events asynchronously

Operations and tasks expose an events() async iterator:

operation = manager.provision("analysis", spec)

async for event in operation.events():
    render_activity(event)

environment = await operation

The iterator replays available history by default and ends after the terminal event.

Cancellation

If the application cancels a coroutine that is awaiting an operation or task, Wetlands requests cancellation of the underlying work. It waits for mandatory process and transfer cleanup before propagating asyncio.CancelledError.

This may make coroutine cancellation take longer than an ordinary in-process awaitable. The delay prevents cleanup from continuing invisibly after the caller sees cancellation as complete.