diff --git a/fastcore/aio.py b/fastcore/aio.py index 18dded38..7460f4a4 100644 --- a/fastcore/aio.py +++ b/fastcore/aio.py @@ -144,7 +144,7 @@ def _f(*args, **kwargs): return CachedAwaitable(func(*args, **kwargs)) def is_async_callable(obj): "Check if `obj` is an async callable, handling `partial` wrappers and callable instances" # Implementation from Starlette; Copyright © 2018, Encode OSS Ltd. - from asyncio import iscoroutinefunction + from inspect import iscoroutinefunction while isinstance(obj, partial): obj = obj.func return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__)) diff --git a/fastcore/parallel.py b/fastcore/parallel.py index 9c6e235b..05f2a662 100644 --- a/fastcore/parallel.py +++ b/fastcore/parallel.py @@ -156,12 +156,12 @@ def _add_one(x, a=1): async def parallel_async_gen(f, items, *args, n_workers=16, pause=0, timeout=None, return_exceptions=False, cancel_on_exit=True, **kwargs): "Yield `(index,result)` pairs as `f` applied to each of `items` completes, in completion order" - import asyncio + import asyncio, inspect semaphore = asyncio.Semaphore(n_workers) async def limited_task(i, item): if pause: await asyncio.sleep(i * pause) async with semaphore: - coro = f(item, *args, **kwargs) if asyncio.iscoroutinefunction(f) else asyncio.to_thread(f, item, *args, **kwargs) + coro = f(item, *args, **kwargs) if inspect.iscoroutinefunction(f) else asyncio.to_thread(f, item, *args, **kwargs) try: return i, (await asyncio.wait_for(coro, timeout) if timeout else await coro) except Exception as e: if return_exceptions: return i, e diff --git a/fastcore/xtras.py b/fastcore/xtras.py index 6c11fc44..e72707e4 100644 --- a/fastcore/xtras.py +++ b/fastcore/xtras.py @@ -1232,7 +1232,7 @@ def __iter__(self): # %% ../nbs/03_xtras.ipynb #d2b4fe09 def flexicache(*funcs, maxsize=128): "Like `lru_cache`, but customisable with policy `funcs`" - import asyncio + import asyncio, inspect from collections import OrderedDict wants = [getattr(f, 'needs_args', False) for f in funcs] def _f(func): @@ -1262,7 +1262,7 @@ def wrapper(*args, **kwargs): async def async_wrapper(*args, **kwargs): return await _cache_logic(f"{args} // {kwargs}", lambda: asyncio.ensure_future(func(*args, **kwargs)), args, kwargs) - return async_wrapper if asyncio.iscoroutinefunction(func) else wrapper + return async_wrapper if inspect.iscoroutinefunction(func) else wrapper return _f # %% ../nbs/03_xtras.ipynb #cf131274 diff --git a/nbs/03_xtras.ipynb b/nbs/03_xtras.ipynb index 725baade..c34eeafb 100644 --- a/nbs/03_xtras.ipynb +++ b/nbs/03_xtras.ipynb @@ -5881,7 +5881,7 @@ "#| export\n", "def flexicache(*funcs, maxsize=128):\n", " \"Like `lru_cache`, but customisable with policy `funcs`\"\n", - " import asyncio\n", + " import asyncio, inspect\n", " from collections import OrderedDict\n", " wants = [getattr(f, 'needs_args', False) for f in funcs]\n", " def _f(func):\n", @@ -5911,7 +5911,7 @@ " async def async_wrapper(*args, **kwargs):\n", " return await _cache_logic(f\"{args} // {kwargs}\", lambda: asyncio.ensure_future(func(*args, **kwargs)), args, kwargs)\n", "\n", - " return async_wrapper if asyncio.iscoroutinefunction(func) else wrapper\n", + " return async_wrapper if inspect.iscoroutinefunction(func) else wrapper\n", " return _f" ] }, diff --git a/nbs/03a_parallel.ipynb b/nbs/03a_parallel.ipynb index 34367386..928a0923 100644 --- a/nbs/03a_parallel.ipynb +++ b/nbs/03a_parallel.ipynb @@ -588,12 +588,12 @@ "async def parallel_async_gen(f, items, *args, n_workers=16, pause=0,\n", " timeout=None, return_exceptions=False, cancel_on_exit=True, **kwargs):\n", " \"Yield `(index,result)` pairs as `f` applied to each of `items` completes, in completion order\"\n", - " import asyncio\n", + " import asyncio, inspect\n", " semaphore = asyncio.Semaphore(n_workers)\n", " async def limited_task(i, item):\n", " if pause: await asyncio.sleep(i * pause)\n", " async with semaphore:\n", - " coro = f(item, *args, **kwargs) if asyncio.iscoroutinefunction(f) else asyncio.to_thread(f, item, *args, **kwargs)\n", + " coro = f(item, *args, **kwargs) if inspect.iscoroutinefunction(f) else asyncio.to_thread(f, item, *args, **kwargs)\n", " try: return i, (await asyncio.wait_for(coro, timeout) if timeout else await coro)\n", " except Exception as e:\n", " if return_exceptions: return i, e\n", diff --git a/nbs/03c_aio.ipynb b/nbs/03c_aio.ipynb index d896a9b7..98893a79 100644 --- a/nbs/03c_aio.ipynb +++ b/nbs/03c_aio.ipynb @@ -299,7 +299,8 @@ " time.sleep(0.25)\n", " return x*2\n", "\n", - "assert asyncio.iscoroutinefunction(_slow)\n", + "import inspect\n", + "assert inspect.iscoroutinefunction(_slow)\n", "start = time.time()\n", "test_eq(await asyncio.gather(_slow(1), _slow(2)), [2,4])\n", "elapsed = time.time()-start\n", @@ -550,7 +551,7 @@ "def is_async_callable(obj):\n", " \"Check if `obj` is an async callable, handling `partial` wrappers and callable instances\"\n", " # Implementation from Starlette; Copyright © 2018, Encode OSS Ltd.\n", - " from asyncio import iscoroutinefunction\n", + " from inspect import iscoroutinefunction\n", " while isinstance(obj, partial): obj = obj.func\n", " return iscoroutinefunction(obj) or (callable(obj) and iscoroutinefunction(obj.__call__))" ] @@ -560,7 +561,7 @@ "id": "d3a3c5f3", "metadata": {}, "source": [ - "`is_async_callable` detects whether an object can be called asynchronously. It goes beyond `asyncio.iscoroutinefunction` by also handling `functools.partial`-wrapped async functions (unwrapping through any number of layers) and callable objects whose `__call__` method is a coroutine. The implementation is thanks to [Starlette](https://github.com/encode/starlette)." + "`is_async_callable` detects whether an object can be called asynchronously. It goes beyond `inspect.iscoroutinefunction` by also handling `functools.partial`-wrapped async functions (unwrapping through any number of layers) and callable objects whose `__call__` method is a coroutine. The implementation is thanks to [Starlette](https://github.com/encode/starlette)." ] }, {