Output and prompts from multiple threads

class click_threading.UiWorker

A worker-queue system to manage and synchronize output and prompts from other threads.

>>> import click
>>> from click_threading import UiWorker, Thread, get_ui_worker
>>> ui = UiWorker()  # on main thread
>>> def target():
...     click.echo("Hello world!")
...     get_ui_worker().shutdown()
...
>>>
>>> @click.command()
... def cli():
...     with ui.patch_click():
...         t = Thread(target=target)
...         t.start()
...         ui.run()
>>> runner = click.testing.CliRunner()
>>> result = runner.invoke(cli, [])
>>> assert result.output.strip() == 'Hello world!'

Using this class instead of just spawning threads brings a few advantages:

  • If one thread prompts for input, other output from other threads is queued until the click.prompt() call returns.
  • If you call echo with a multiline-string, it is guaranteed that this string is not interleaved with other output.

Disadvantages:

  • The main thread is used for the output (using any other thread produces weird behavior with interrupts). ui.run() in the above example blocks until ui.shutdown() is called.