Skip to content

Async Commands

cmd2 is built on top of the Python Standard Library's cmd module, which is inherently synchronous. This means that do_* command methods are expected to be synchronous functions.

However, you can still integrate asynchronous code (using asyncio and async/await) into your cmd2 application by running an asyncio event loop in a background thread and bridging calls to it.

The with_async_loop Decorator

A clean way to handle this is to define a decorator that wraps your async def commands. This decorator handles:

  1. Starting a background thread with an asyncio loop (if not already running).
  2. Submitting the command's coroutine to that loop.
  3. Waiting for the result (synchronously) so that the cmd2 interface behaves as expected (blocking until the command completes).

Example Implementation

Here is an example of how to implement such a decorator and use it in your application.

import asyncio
import functools
import threading
from typing import Any, Callable
import cmd2

# Global event loop and lock
_event_loop = None
_event_lock = threading.Lock()

def _get_event_loop() -> asyncio.AbstractEventLoop:
    """Get or create the background event loop."""
    global _event_loop

    if _event_loop is None:
        with _event_lock:
            if _event_loop is None:
                _event_loop = asyncio.new_event_loop()
                thread = threading.Thread(
                    target=_event_loop.run_forever,
                    name='Async Runner',
                    daemon=True,
                )
                thread.start()
    return _event_loop

def with_async_loop(func: Callable[..., Any]) -> Callable[..., Any]:
    """Decorator to run a command method asynchronously in a background thread."""
    @functools.wraps(func)
    def wrapper(self: cmd2.Cmd, *args: Any, **kwargs: Any) -> Any:
        loop = _get_event_loop()
        coro = func(self, *args, **kwargs)
        future = asyncio.run_coroutine_threadsafe(coro, loop)
        return future.result()
    return wrapper

class AsyncApp(cmd2.Cmd):
    @with_async_loop
    async def do_my_async(self, _: cmd2.Statement) -> None:
        self.poutput("Starting async work...")
        await asyncio.sleep(1.0)
        self.poutput("Async work complete!")

See Also