Skip to content

Add async function extensions #32

@binyamin555

Description

@binyamin555

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 result

Then, 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.

Metadata

Metadata

Assignees

No one assigned

    Labels

    enhancementNew feature or request

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions