Add Caller.get_runner. (#126)

Author: fleming79

src/async_kernel/caller.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import asyncio
 import contextlib
 import contextvars
 import functools
@@ -11,6 +12,7 @@
 import weakref
 from collections import deque
 from collections.abc import AsyncGenerator, Awaitable, Callable, Generator
+from types import CoroutineType
 from typing import TYPE_CHECKING, Any, ClassVar, Literal, Self, cast, overload
 
 import anyio
@@ -556,6 +558,22 @@ def stopped(self) -> bool:
         "Returns  `True` if the caller is stopped."
         return self._stopped
 
+    def get_runner(self, *, started: Callable[[], None] | None = None) -> CoroutineType[Any, Any, None]:
+        """A convenience method to run the caller.
+
+        !!! tip
+
+            See [async_kernel.caller.Caller.get_instance][] for a usage example.
+        """
+        if self.running or self.stopped:
+            raise RuntimeError
+
+        async def runner() -> None:
+            async with self:
+                await anyio.sleep_forever()
+
+        return runner()
+
     def stop(self, *, force=False) -> None:
         """
         Stop the caller, cancelling all pending tasks and close the thread.
@@ -751,8 +769,17 @@ def get_instance(cls, name: str | None = "MainThread", *, create: bool = False)
         for thread in cls._instances:
             if thread.name == name:
                 return cls._instances[thread]
-        if create:
-            return cls.start_new(name=name)
+        if name == "MainThread":
+            if threading.current_thread() is threading.main_thread():
+                if (backend := sniffio.current_async_library()) == Backend.asyncio:
+                    inst = cls(create=True)
+                    inst._task = asyncio.create_task(inst.get_runner())  # pyright: ignore[reportAttributeAccessIssue]
+                    return inst
+                msg = f"Starting a caller for the MainThread is not supported for {backend=}"
+                raise RuntimeError(msg)
+        else:
+            if create is True:
+                return cls.start_new(name=name)
         msg = f"A Caller was not found for {name=}."
         raise RuntimeError(msg)
 

tests/test_caller.py

@@ -349,6 +349,23 @@ async def test_get_instance_no_instance(self, anyio_backend):
         with pytest.raises(RuntimeError):
             Caller.get_instance(None, create=False)
 
+    async def test_get_instance_get_runner(self, anyio_backend):
+        if anyio_backend == Backend.trio:
+            with pytest.raises(RuntimeError):
+                Caller.get_instance()
+            return
+        caller = Caller.get_instance()
+        try:
+            await caller.call_soon(anyio.sleep, 0.01)
+        finally:
+            caller.stop()
+
+    async def test_get_runner_error(self):
+        caller = Caller(create=True)
+        caller.stop()
+        with pytest.raises(RuntimeError):
+            caller.get_runner()  # pyright: ignore[reportUnusedCoroutine]
+
     @pytest.mark.parametrize("mode", ["restricted", "surge"])
     async def test_as_completed(self, anyio_backend, mode: Literal["restricted", "surge"], mocker):
         mocker.patch.object(Caller, "MAX_IDLE_POOL_INSTANCES", new=2)