Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion fastcore/aio.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__))

Expand Down
4 changes: 2 additions & 2 deletions fastcore/parallel.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions fastcore/xtras.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down Expand Up @@ -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
Expand Down
4 changes: 2 additions & 2 deletions nbs/03_xtras.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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"
]
},
Expand Down
4 changes: 2 additions & 2 deletions nbs/03a_parallel.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down
7 changes: 4 additions & 3 deletions nbs/03c_aio.ipynb
Original file line number Diff line number Diff line change
Expand Up @@ -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",
Expand Down Expand Up @@ -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__))"
]
Expand All @@ -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)."
]
},
{
Expand Down
Loading