-
Notifications
You must be signed in to change notification settings - Fork 0
Open
Description
When passing firstresult=True to a hookspec marker, the HookCaller will only return 1 result (coroutine) instead of a list of results but asyncio.gather expect a list of "tasks" and not a single coroutine
Here's a minimally reproducible example to illustrate the issue.
import apluggy
import asyncio
hookspec = apluggy.HookspecMarker("my-project")
hookimpl = apluggy.HookimplMarker("my-project")
class Hookspec:
@hookspec(firstresult=True)
async def afunc(self, x, y): ...
class Plugin:
@hookimpl
async def afunc(self, x, y):
return x + y
async def main():
pm = apluggy.PluginManager("my-project")
pm.add_hookspecs(Hookspec())
pm.register(Plugin)
print(await pm.ahook.afunc(x=1, y=2))
asyncio.run(main())The code above fails with TypeError: asyncio.tasks.gather() argument after * must be an iterable, not coroutine
but changing the call function to actually check whether calling hook(*args, **kwargs) returns a list and acting accordingly if it doesn't fixes the issue.
class AHook:
def __init__(self, pm: PluginManager_) -> None:
self.pm = pm
def __getattr__(self, name: str) -> Callable[..., Coroutine[Any, Any, list]]:
async def call(*args: Any, **kwargs: Any) -> list:
hook: HookCaller = getattr(self.pm.hook, name)
coros: list[asyncio.Future] = hook(*args, **kwargs)
if not isinstance(coros, Coroutine):
return None
if not isinstance(coros, list): # Added an isinstance check to see whether a list or a single element is returned
return await coros
return await asyncio.gather(*coros)
return callAfter this small change the code works as expected:
print(await pm.ahook.afunc(x=1, y=2))
>>> 3psychotechnik
Metadata
Metadata
Assignees
Labels
No labels