Skip to content

Easy Async

py_simple.easy_async

easy_async is built to simplify asynchronous code execution.

EasyAsyncError

Bases: Exception

Raised when a py_simple async function fails to complete.

Wraps the underlying error (an exception raised inside one of the functions being run, a thread pool failure, etc.) so py_simple functions can fail with one consistent, easy-to-read exception instead of a random builtin one.

Parameters:

Name Type Description Default
message str

Human-readable description of what went wrong.

 required

run_after_delay(func, delay, *args) async

Waits asynchronously before running a function.

Raises EasyAsyncError if the function raises an exception after the delay finishes.

Parameters:

Name Type Description Default
func callable

The function to execute.

 required
delay float

Number of seconds to wait before running the function.

 required
*args

Positional arguments to pass to the function.

 ()

Returns:

Name Type Description
tuple tuple

A tuple containing (func.__name__, result).
Example 
import asyncio
from py_simple import run_after_delay

def greet(name):
    return f"Hello, {name}!"

async def main():
    return await run_after_delay(greet, 1.0, "Sam")

asyncio.run(main())  # -> ("greet", "Hello, Sam!")

import asyncio

def greet(name):
    return f"Hello, {name}!"

async def main():
    await asyncio.sleep(1.0)
    loop = asyncio.get_running_loop()
    result = await loop.run_in_executor(None, greet, "Sam")
    return ("greet", result)

run_at_the_same_time_no_params(functions) async

Runs multiple zero-argument functions at the same time and returns their results.

Raises EasyAsyncError if any function raises an exception while running.

Parameters:

Name Type Description Default
functions list

Functions to run, each taking no arguments.

 required

Returns:

Name Type Description
list list

A list of (name, result) tuples, one per function.
Example 
import asyncio
from py_simple import run_at_the_same_time_no_params

def add():
    return 1 + 1

def sub():
    return 4 - 2

async def main():
    return await run_at_the_same_time_no_params([add, sub])

asyncio.run(main())  # -> [("add", 2), ("sub", 2)]

import asyncio

def add():
    return 1 + 1

def sub():
    return 4 - 2

async def main():
    loop = asyncio.get_running_loop()
    functions = [add, sub]
    tasks = [loop.run_in_executor(None, f) for f in functions]
    results = await asyncio.gather(*tasks)
    return list(zip((f.__name__ for f in functions), results))

asyncio.run(main())

run_at_the_same_time_with_params(functions_and_args) async

Runs multiple functions at the same time, each with its own arguments, and returns their results.

Raises EasyAsyncError if any function raises an exception while running.

Parameters:

Name Type Description Default
functions_and_args list[tuple]

Functions to run, each given as a tuple where the first item is the function and the remaining items are the positional arguments to call it with, e.g. (func, arg1, arg2).

 required

Returns:

Name Type Description
list list

A list of (name, result) tuples, one per function.
Example 
import asyncio
from py_simple import run_at_the_same_time_with_params

def add(a, b):
    return a + b

def sub(a, b):
    return a - b

async def main():
    return await run_at_the_same_time_with_params([
        (add, 1, 1),
        (sub, 4, 2),
    ])

asyncio.run(main())  # -> [("add", 2), ("sub", 2)]

import asyncio

def add(a, b):
    return a + b

def sub(a, b):
    return a - b

async def main():
    loop = asyncio.get_running_loop()
    functions_and_args = [(add, 1, 1), (sub, 4, 2)]
    tasks = [
        loop.run_in_executor(None, item[0], *item[1:])
        for item in functions_and_args
    ]
    results = await asyncio.gather(*tasks)
    names = [item[0].__name__ for item in functions_and_args]
    return list(zip(names, results))

asyncio.run(main())

run_concurrent_map(func, items) async

Applies a function to a list of items concurrently and returns the results in the original order.

Raises EasyAsyncError if any function call raises an exception.

Parameters:

Name Type Description Default
func callable

The function to call on each item.

 required
items list

The list of items to pass one by one into the function.

 required

Returns:

Name Type Description
list list

The list of return values in the same order as items.
Example 
import asyncio
from py_simple import run_concurrent_map

def square(n):
    return n * n

async def main():
    return await run_concurrent_map(square, [1, 2, 3, 4])

asyncio.run(main())  # -> [1, 4, 9, 16]

import asyncio

def square(n):
    return n * n

async def main():
    loop = asyncio.get_running_loop()
    items = [1, 2, 3, 4]
    tasks = [loop.run_in_executor(None, square, item) for item in items]
    return await asyncio.gather(*tasks)

asyncio.run(main())

run_periodically(func, interval, times, *args) async

Runs the same function repeatedly on a fixed schedule and returns the results in order.

Waits interval seconds before each run (including the first), runs func in a thread pool so it never blocks the event loop, and collects one result per run. Raises EasyAsyncError if any run raises an exception while running.

Parameters:

Name Type Description Default
func callable

The function to run, each time with the same positional arguments.

 required
interval float

Seconds to wait before each run, including the first one. Must be greater than 0.

 required
times int

How many times to run the function. Must be at least 1.

 required
*args

Positional arguments to pass to func on every run.

 ()

Returns:

Name Type Description
list list

One result per run, in the order the runs finished, e.g. [4, 4, 4] for a function that returns 2 + 2.

Raises:

Type Description
EasyAsyncError

If a run raises an exception, or if interval or times are out of range.
Example 
import asyncio
from py_simple import run_periodically

def add():
    return 2 + 2

async def main():
    return await run_periodically(add, 1.0, 3)

asyncio.run(main())  # -> [4, 4, 4]

import asyncio

def add():
    return 2 + 2

async def main():
    loop = asyncio.get_running_loop()
    results = []
    for _ in range(3):
        await asyncio.sleep(1.0)
        results.append(await loop.run_in_executor(None, add))
    return results

asyncio.run(main())  # -> [4, 4, 4]

run_with_retry(func, attempts, delay, *args) async 

Runs a function repeatedly until it succeeds or all attempts are
exhausted, waiting between attempts.

Raises EasyAsyncError if every attempt fails.

Args:
    func (callable): The function to execute.
    attempts (int): Number of attempts to make before giving up.
    delay (float): Time to wait between failed attempts, in seconds.
    *args: Positional arguments to pass to the function.

Returns:
    tuple: A tuple containing `(func.__name__, result)`.

Example:
    === "The Py_simple Way"
        ```python
        import asyncio
        from py_simple import run_with_retry

        def add(a, b):
            return a + b

        async def main():
            return await run_with_retry(add, 4, 2.0, 2, 3)

        asyncio.run(main())  # -> ("add", 5)
        ```

    === "The Traditional Way"
        ```python
        import asyncio

        def add(a, b):
            return a + b

        async def main():
            loop = asyncio.get_running_loop()
            attempts = 4
            for attempt in range(attempts):
                try:
                    result = await loop.run_in_executor(None, add, 2, 3)
                    return (add.__name__, result)
                except Exception as e:
                    if attempt == attempts - 1:
                        raise EasyAsyncError(f"

ERROR: {e}") from None await asyncio.sleep(2.0)

        asyncio.run(main())
        ```

run_with_timeout(func, timeout, *args) async

Runs a function asynchronously with a timeout limit.

Raises EasyAsyncError if the function times out or raises an exception.

Parameters:

Name Type Description Default
func callable

The function to execute.

 required
timeout float

Maximum time to wait in seconds.

 required
*args

Positional arguments to pass to the function.

 ()

Returns:

Name Type Description
tuple tuple

A tuple containing (func.__name__, result).
Example 
import asyncio
from py_simple import run_with_timeout

def slow_add(a, b):
    return a + b

async def main():
    return await run_with_timeout(slow_add, 2.0, 3, 5)

asyncio.run(main())  # -> ("slow_add", 8)

import asyncio

def slow_add(a, b):
    return a + b

async def main():
    loop = asyncio.get_running_loop()
    result = await asyncio.wait_for(
        loop.run_in_executor(None, slow_add, 3, 5),
        timeout=2.0,
    )
    return (slow_add.__name__, result)

asyncio.run(main())