feat(async): greenlet Submitter and PluginManager.run_async

.pre-commit-config.yaml

@@ -46,4 +46,4 @@ repos:
     -   id: mypy
         files: ^(src/|testing/)
         args: []
-        additional_dependencies: [pytest]
+        additional_dependencies: [pytest, types-greenlet]

changelog/710.feature.rst

@@ -0,0 +1,9 @@
+New async bridge for awaitable hook results:
+``await pm.run_async(lambda: pm.hook.my_hook())`` runs the hook call in a
+greenlet context where awaitable results from hook implementations are
+awaited on the current event loop ("await-me-maybe"). Outside
+``run_async``, awaitable results are passed through unchanged, as before.
+Requires the optional ``greenlet`` dependency, installable via
+``pluggy[async]``. The bridge is a persistent per-manager ``Submitter``;
+async wrapper generators are not auto-awaited (a manual
+``async_generator_to_sync`` helper is provided).

pyproject.toml

@@ -44,9 +44,12 @@ Repository = "https://github.com/pytest-dev/pluggy"
 Issues = "https://github.com/pytest-dev/pluggy/issues"
 Funding = "https://tidelift.com/subscription/pkg/pypi-pluggy"
 
+[project.optional-dependencies]
+async = ["greenlet"]
+
 [dependency-groups]
 dev = ["pre-commit"]
-testing = ["pytest>=8", "pytest-benchmark", "coverage"]
+testing = ["pytest>=8", "pytest-benchmark", "coverage", "greenlet"]
 
 
 [tool.setuptools]

src/pluggy/_async.py

@@ -0,0 +1,173 @@
+"""
+Async support for pluggy using greenlets.
+
+This module provides async functionality for pluggy, allowing hook
+implementations to return awaitable objects that are automatically awaited
+when running in an async context (see :meth:`PluginManager.run_async
+<pluggy.PluginManager.run_async>`).
+"""
+
+from __future__ import annotations
+
+from collections.abc import AsyncGenerator
+from collections.abc import Awaitable
+from collections.abc import Callable
+from collections.abc import Generator
+from typing import Any
+from typing import cast
+from typing import Final
+from typing import TYPE_CHECKING
+from typing import TypeVar
+
+
+_T = TypeVar("_T")
+_Y = TypeVar("_Y")
+_S = TypeVar("_S")
+
+if TYPE_CHECKING:
+    import greenlet
+
+
+#: Sentinel distinguishing "worker never completed" from a legitimate
+#: ``None`` result.
+_UNSET: Final = object()
+
+
+class Submitter:
+    """Bridge between synchronous hook execution and an async event loop.
+
+    While :meth:`run` is active, awaitables passed to :meth:`maybe_submit` or
+    :meth:`require_await` are switched to the awaiting parent greenlet, which
+    awaits them and switches the result back. When inactive,
+    :meth:`maybe_submit` passes awaitables through unchanged
+    ("await-me-maybe").
+    """
+
+    _active_submitter: greenlet.greenlet | None
+
+    def __init__(self) -> None:
+        self._active_submitter = None
+
+    def __repr__(self) -> str:
+        return f"<Submitter active={self._active_submitter is not None}>"
+
+    def maybe_submit(self, coro: Awaitable[_T]) -> _T | Awaitable[_T]:
+        """Await an awaitable if active, else return it unchanged.
+
+        This enables backward compatibility for datasette
+        and https://simonwillison.net/2020/Sep/2/await-me-maybe/
+        """
+        active = self._active_submitter
+        if active is not None:
+            # We're in a greenlet context; switch to the parent with the
+            # awaitable. The parent awaits it and switches back the result.
+            res: _T = active.switch(coro)
+            return res
+        else:
+            return coro
+
+    def require_await(self, coro: Awaitable[_T]) -> _T:
+        """Await an awaitable, raising an error if not in async context.
+
+        Ownership of ``coro`` is taken either way: when inactive it is closed
+        before raising, so callers never leak a coroutine that would later
+        warn about never having been awaited.
+        """
+        active = self._active_submitter
+        if active is not None:
+            res: _T = active.switch(coro)
+            return res
+        else:
+            # Not going to await it, so dispose of it rather than let it be
+            # collected unawaited (a RuntimeWarning raised inside __del__).
+            close = getattr(coro, "close", None)
+            if close is not None:
+                close()
+            raise RuntimeError("require_await called outside of async context")
+
+    async def run(self, sync_func: Callable[[], _T]) -> _T:
+        """Run a synchronous function in a worker greenlet with async support.
+
+        Awaitables submitted by hook implementations during the call are
+        awaited on this coroutine's event loop.
+        """
+        try:
+            import greenlet
+        except ImportError:
+            raise RuntimeError("greenlet is required for async support") from None
+
+        if self._active_submitter is not None:
+            raise RuntimeError("Submitter is already active")
+
+        main_greenlet = greenlet.getcurrent()
+        result: object = _UNSET
+        exception: BaseException | None = None
+
+        def greenlet_func() -> None:
+            nonlocal result, exception
+            try:
+                result = sync_func()
+            except BaseException as e:
+                exception = e
+
+        worker_greenlet = greenlet.greenlet(greenlet_func)
+        # Let maybe_submit/require_await switch back to this greenlet.
+        self._active_submitter = main_greenlet
+        try:
+            # Run the worker; every switch back carries an awaitable to
+            # process, until the worker finishes (switching back None).
+            awaitable = worker_greenlet.switch()
+            while awaitable is not None:
+                try:
+                    awaited_result = await awaitable
+                except BaseException as e:
+                    # Raise at the submission site inside the worker.
+                    awaitable = worker_greenlet.throw(e)
+                else:
+                    awaitable = worker_greenlet.switch(awaited_result)
+        finally:
+            self._active_submitter = None
+
+        if exception is not None:
+            raise exception
+        assert result is not _UNSET, "worker greenlet did not complete"
+        return cast(_T, result)
+
+
+def async_generator_to_sync(
+    async_gen: AsyncGenerator[_Y, _S], submitter: Submitter
+) -> Generator[_Y, _S, None]:
+    """Convert an async generator to a sync generator using a `Submitter`.
+
+    This helper allows wrapper implementations to use async generators while
+    maintaining compatibility with the sync generator interface expected by
+    the hook system. The submitter must be active (i.e. the generator must be
+    consumed under :meth:`Submitter.run`).
+    """
+    try:
+        # Start the async generator.
+        value = submitter.require_await(async_gen.__anext__())
+
+        while True:
+            try:
+                sent_value = yield value
+                try:
+                    value = submitter.require_await(async_gen.asend(sent_value))
+                except StopAsyncIteration:
+                    return
+
+            except GeneratorExit:
+                # Generator is being closed; close the async generator too.
+                submitter.require_await(cast(Awaitable[Any], async_gen.aclose()))
+                raise
+
+            except BaseException as exc:
+                # Exception was thrown into the generator; forward it into
+                # the async generator.
+                try:
+                    value = submitter.require_await(async_gen.athrow(exc))
+                except StopAsyncIteration:
+                    return
+
+    except StopAsyncIteration:
+        return

src/pluggy/_caller.py

@@ -19,6 +19,7 @@
 from typing import TypeAlias
 from typing import TypeVar
 
+from ._async import Submitter
 from ._config import HookimplConfiguration
 from ._config import hookspec_config_from_mapping
 from ._config import HookspecConfiguration
@@ -31,7 +32,14 @@
 
 
 _HookExec: TypeAlias = Callable[
-    [str, Sequence[NormalImpl], Sequence[WrapperImpl], Mapping[str, object], bool],
+    [
+        str,
+        Sequence[NormalImpl],
+        Sequence[WrapperImpl],
+        Mapping[str, object],
+        bool,
+        Submitter,
+    ],
     "object | list[object]",
 ]
 
@@ -153,6 +161,7 @@ class NormalHookCaller:
     """A caller of all registered implementations of a hook specification."""
 
     __slots__ = (
+        "_async_submitter",
         "_hookexec",
         "_normal_hookimpls",
         "_wrapper_hookimpls",
@@ -166,11 +175,13 @@ def __init__(
         hook_execute: _HookExec,
         specmodule_or_class: _Namespace | None = None,
         spec_config: HookspecConfiguration | None = None,
+        async_submitter: Submitter | None = None,
     ) -> None:
         """:meta private:"""
         #: Name of the hook getting called.
         self.name: Final = name
         self._hookexec: Final = hook_execute
+        self._async_submitter: Final = async_submitter or Submitter()
         # Split hook implementations into two lists for simpler management:
         # Normal hooks: [trylast, normal, tryfirst]
         # Wrapper hooks: [trylast, normal, tryfirst]
@@ -262,6 +273,7 @@ def __call__(self, **kwargs: object) -> Any:
             self._wrapper_hookimpls.copy(),
             call_kwargs,
             firstresult,
+            self._async_submitter,
         )
 
     def call_historic(
@@ -297,6 +309,7 @@ def call_extra(
             self._wrapper_hookimpls.copy(),
             kwargs,
             firstresult,
+            self._async_submitter,
         )
 
     def _maybe_apply_history(self, method: HookImpl) -> None:
@@ -316,6 +329,7 @@ class HistoricHookCaller:
     """
 
     __slots__ = (
+        "_async_submitter",
         "_call_history",
         "_hookexec",
         "_hookimpls",
@@ -331,12 +345,14 @@ def __init__(
         hook_execute: _HookExec,
         specmodule_or_class: _Namespace,
         spec_config: HookspecConfiguration,
+        async_submitter: Submitter | None = None,
     ) -> None:
         """:meta private:"""
         assert spec_config.historic, "HistoricHookCaller requires historic=True"
         #: Name of the hook getting called.
         self.name: Final = name
         self._hookexec: Final = hook_execute
+        self._async_submitter: Final = async_submitter or Submitter()
         # The hookimpls list for historic hooks (no wrappers supported).
         self._hookimpls: Final[list[NormalImpl]] = []
         self._call_history: Final[_CallHistory] = []
@@ -410,7 +426,14 @@ def call_historic(
         # Historizing hooks don't return results.
         # Remember firstresult isn't compatible with historic.
         # Copy because plugins may register other plugins during iteration (#438).
-        res = self._hookexec(self.name, self._hookimpls.copy(), [], kwargs, False)
+        res = self._hookexec(
+            self.name,
+            self._hookimpls.copy(),
+            [],
+            kwargs,
+            False,
+            self._async_submitter,
+        )
         if result_callback is None:
             return
         if isinstance(res, list):
@@ -429,7 +452,9 @@ def _maybe_apply_history(self, method: HookImpl) -> None:
         """Apply call history to a new hookimpl."""
         assert isinstance(method, NormalImpl)
         for kwargs, result_callback in self._call_history:
-            res = self._hookexec(self.name, [method], [], kwargs, False)
+            res = self._hookexec(
+                self.name, [method], [], kwargs, False, self._async_submitter
+            )
             if res and result_callback is not None:
                 # XXX: remember firstresult isn't compat with historic
                 assert isinstance(res, list)
@@ -511,6 +536,7 @@ def __call__(self, **kwargs: object) -> Any:
             self._get_filtered(orig._wrapper_hookimpls),
             call_kwargs,
             firstresult,
+            orig._async_submitter,
         )
 
     def call_historic(
@@ -532,7 +558,12 @@ def call_historic(
         # History is shared with the original caller.
         orig._call_history.append((kwargs, result_callback))
         res = orig._hookexec(
-            self.name, self._get_filtered(orig._hookimpls), [], kwargs, False
+            self.name,
+            self._get_filtered(orig._hookimpls),
+            [],
+            kwargs,
+            False,
+            orig._async_submitter,
         )
         if result_callback is None:
             return
@@ -568,6 +599,7 @@ def call_extra(
             self._get_filtered(orig._wrapper_hookimpls),
             kwargs,
             firstresult,
+            orig._async_submitter,
         )
 
     def __repr__(self) -> str:

src/pluggy/_execution.py

@@ -4,6 +4,7 @@
 
 from __future__ import annotations
 
+from collections.abc import Awaitable
 from collections.abc import Generator
 from collections.abc import Mapping
 from collections.abc import Sequence
@@ -20,6 +21,10 @@
 from ._warnings import PluggyTeardownRaisedWarning
 
 
+if TYPE_CHECKING:
+    from ._async import Submitter
+
+
 # Need to distinguish between old- and new-style hook wrappers.
 # Wrapping with a tuple is the fastest type-safe way I found to do it.
 Teardown: TypeAlias = Generator[None, object, object]
@@ -93,6 +98,7 @@ def _multicall(
     wrapper_impls: Sequence[WrapperImpl],
     caller_kwargs: Mapping[str, object],
     firstresult: bool,
+    async_submitter: Submitter,
 ) -> object | list[object]:
     """Execute a call into multiple python functions/methods and return the
     result(s).
@@ -127,6 +133,10 @@ def _multicall(
             else:
                 res = normal_impl.function(*args, **kwargs)
             if res is not None:
+                # Awaitable results are awaited when a Submitter is active
+                # (await-me-maybe), otherwise passed through unchanged.
+                if isinstance(res, Awaitable):
+                    res = async_submitter.maybe_submit(res)
                 results.append(res)
                 if firstresult:  # halt further impl calls
                     break