-
Notifications
You must be signed in to change notification settings - Fork 2
Description
We can (optionally) extend coroutine objects with the following extensions:
def then(Awaitable[T], Callable[[T], Awaitable[U] | U]) -> Awaitable[U]
def catch(Awaitable[T], Callable[[E], Awaitable[U] | U], *, exception: type[E] = Excpetion) -> Awaitable[T | U]The then function basically maps the result of the first awaitable via an optionally async function. If the function is async, it is awaited in the context of the wrapped awaitable.
The catch function catches an exception of the given type and the passed function is called with the caught exception.
If no exception was raised inside the wrapped awaitable, the function will not be called.
The passed function can optionally return a value to be returned in case of an error.
The passed function can be either sync or async. If it's async, it is awaited in the context of the wrapped awaitable.
Implementation Suggestion
First, create 2 wrapping functions:
from inspect import iscoroutinefunction as is_async
async def then(awaitable, fn):
result = fn(await awaitable)
if is_async(fn):
return await result
return result
async def catch(awaitable, fn, *, exception=Exception):
try:
return await awaitable
except exception as e:
result = fn(e)
if is_async(fn):
return await result
return resultThen, the extension methods will call these respectively, passing self as the awaitable and the other arguments as is.
A PR implementing this should also consider updating README.md to include the added built-in extension.