1
import asyncio
2
import functools
3
import threading
4
from typing import Any, Callable
5
import cmd2
6

7
# Global event loop and lock
8
_event_loop = None
9
_event_lock = threading.Lock()
10

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

15
    if _event_loop is None:
16
        with _event_lock:
17
            if _event_loop is None:
18
                _event_loop = asyncio.new_event_loop()
19
                thread = threading.Thread(
20
                    target=_event_loop.run_forever,
21
                    name='Async Runner',
22
                    daemon=True,
23
                )
24
                thread.start()
25
    return _event_loop
26

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

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