click-async-plugins

This is a proof-of-concept of a Python asyncio plugin architecture based on click.

Instead of writing functions that make up sub-commands, you write asynchronous "lifespan functions" (AsyncGenerator), like this one:

@cli_core.plugin_command
@click.option("--sleep", type=click.FloatRange(min=0.01), default=1)
async def myplugin(sleep: float) -> PluginLifespan:

    # code to set things up goes here

    async def long_running_task(*, sleep: float):
        # task initialisation can happen here

        try:
            while True:
                await asyncio.sleep(sleep)

        finally:
            # code to clean up the task goes here
            pass

    yield long_running_task(sleep=sleep)

    # code to tear things down goes here

Multiple such plugins can be defined/added to core (see the demo code). These plugins will all have their setup code called in turn. If, after setup, a plugin yields a coroutine (e.g. a long-running task), this task is scheduled with the main event loop, but this is optional, and tasks that yield nothing (None) will just sleep until program termination. Upon termination, the plugins' teardown code is invoked (in reverse order).

Here's what the demo code logs to the console. Two plugins are invoked. The first counts down from 3 each second, and notifies subscribers of each number. The second plugin — "echo" — just listens for updates from the "countdown" task and echoes them.

$ python demo.py countdown --from 3 echo --immediately
DEBUG:root:Setting up task for 'echo'
DEBUG:root:Setting up task for 'countdown'
DEBUG:root:Scheduling task for 'echo'
DEBUG:root:Waiting for update to 'countdown'…
DEBUG:root:Scheduling task for 'countdown'
INFO:root:Counting down… 3
DEBUG:root:Notifying subscribers of update to 'countdown'…
INFO:root:Countdown currently at 3
DEBUG:root:Waiting for update to 'countdown'…
INFO:root:Counting down… 2
DEBUG:root:Notifying subscribers of update to 'countdown'…
INFO:root:Countdown currently at 2
DEBUG:root:Waiting for update to 'countdown'…
INFO:root:Counting down… 1
DEBUG:root:Notifying subscribers of update to 'countdown'…
INFO:root:Countdown currently at 1
DEBUG:root:Waiting for update to 'countdown'…
INFO:root:Finished counting down
^C
DEBUG:root:Task for 'echo' cancelled
DEBUG:root:Terminating…
DEBUG:root:Lifespan over for countdown
DEBUG:root:Lifespan over for echo
DEBUG:root:Finished.

I hope you get the idea. If you need more input, take a look at look at tptools' tpsrv CLI, which I was developing when I factored out this code.

There's also a "debug" plugin included, which allows interaction with the CLI as its running via key presses. Hit ? to get an overview of commands available.

Looking forward to your feedback.

Oh, and if someone wanted to turn this into a proper package with tests and everything, I think it could be published to pip/pypy. I need to stop shaving this yak now, though.

© 2025 martin f. krafft <[email protected]>