coroutines 0.2.0

This is a small package which provides tools to run async functions and generators without an asyncio event loop.

The coroutines module provides functions such as coroutines.run(), coroutines.gather(), and coroutines.sleep() that work just like their asyncio counterparts, but without scheduling any tasks in an external event loop. For example, the following code was adapted from the asyncio documentation:

import coroutines

async def factorial(name, number):
    f = 1
    for i in range(2, number + 1):
        print(f"Task {name}: Compute factorial({number}), currently i={i}...")
        await coroutines.sleep()  # CHANGED: no argument
        f *= i
    print(f"Task {name}: factorial({number}) = {f}")
    return f

async def main():
    # Schedule three calls *concurrently*:
    L = await coroutines.gather(
        factorial("A", 2),
        factorial("B", 3),
        factorial("C", 4),
    )
    print(L)

coroutines.run(main())

# Expected output:
#
#     Task A: Compute factorial(2), currently i=2...
#     Task B: Compute factorial(3), currently i=2...
#     Task C: Compute factorial(4), currently i=2...
#     Task A: factorial(2) = 2
#     Task B: Compute factorial(3), currently i=3...
#     Task C: Compute factorial(4), currently i=3...
#     Task B: factorial(3) = 6
#     Task C: Compute factorial(4), currently i=4...
#     Task C: factorial(4) = 24
#     [2, 6, 24]

The example produces the same result as the asyncio code by simply calling, suspending, and resuming the coroutines until they are completed. Practically, the only difference between the coroutines and asyncio examples is that coroutines.sleep() does not take an argument. Because there is no external event loop, a call to coroutines.sleep() cannot suspend the current chain of coroutines for a fixed amount of time, but only until it is resumed at the next iteration.

coroutines.run(coro) #

coroutine coroutines.sleep() #

coroutine coroutines.gather(*coros, close_on_error=True) #

coroutine coroutines.awaitable(obj=None) #

coroutine coroutines.aiterable(iterable) #

coroutine coroutines.arange(stop) #
coroutine coroutines.arange(start, stop[, step])