async-node 0.1.0

Add your description here

AsyncNode

AsyncNode is a Python asynchronous computation wrapper that allows you to create, chain, combine, and execute asynchronous computations flexibly, optionally using executors for CPU-bound or IO-bound tasks. It supports wrapping synchronous and asynchronous computations, chaining operations using map, run, consume, combining multiple AsyncNode instances, handling exceptions synchronously or asynchronously, and lazy evaluation with caching. Schedulers is a centralized factory for executors that provides singleton-style access, meaning each executor instance (IO-bound, CPU-bound, or single-threaded) is created only once and reused throughout your application. Executors are lazily instantiated when first accessed.

You can create nodes from values, synchronous suppliers, or synchronous runnables, for example:

from async_execution.async_node import AsyncNode

node1 = AsyncNode.from_value(10)
node2 = AsyncNode.from_supplier(lambda: 5)
node3 = AsyncNode.from_runnable(lambda: print("Task done"))

Nodes can be transformed using synchronous mapping functions:

node4 = node1.map(lambda x: x * 2)

or asynchronous mapping functions:

async def async_double(x):
    await asyncio.sleep(1)
    return x * 2

node5 = node2.map_async(async_double)

Side-effects can be run synchronously or asynchronously:

node1.run(lambda: print("Value computed"))

async def async_print():
    await asyncio.sleep(1)
    print("Async task completed")

node2.run_async(async_print)

Nodes can be combined using synchronous functions:

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

combined_node = node1.combine(node2, combine_function=combine_sum)

or asynchronous functions:

async def async_combine(a, b):
    await asyncio.sleep(1)
    return a + b

combined_async_node = node1.combine_async(node2, combine_function=async_combine)

Values can be consumed synchronously or asynchronously:

node1.consume(lambda x: print(f"Consumed value: {x}"))

async def async_consume(x):
    await asyncio.sleep(1)
    print(f"Consumed async: {x}")

node2.consume_async(async_consume)

Exceptions can be handled synchronously or asynchronously:

def handle_exception(e):
    print(f"Exception occurred: {e}")
    return 0

safe_node = node1.exceptionally(handle_exception)

async def async_handle_exception(e):
    print(f"Async exception: {e}")
    return 0

safe_node_async = node2.exceptionally_async(async_handle_exception)

Executors can be used via Schedulers, and they are singleton, so each call returns the same instance:

from async_execution.executors.schedulers import Schedulers

io_executor = Schedulers.io()            # singleton IO-bound executor
cpu_executor = Schedulers.computation()  # singleton CPU-bound executor
single_executor = Schedulers.single()    # singleton single-thread executor

node_with_executor = node1.on(cpu_executor)
node_on_main = node_with_executor.on_main_thread()

Worker timeouts can be configured before executors are created:

Schedulers.set_workers_timeout(30)  # seconds

Finally, results are retrieved asynchronously:

import asyncio

async def main():
    result = await node1.get()
    print(result)

asyncio.run(main())

AsyncNode makes it easy to manage asynchronous workflows in Python, supporting both synchronous and asynchronous functions, chaining, combination, exception handling, and optional executor usage, with lazy evaluation and caching for efficiency. Schedulers simplifies managing and reusing singleton executors for IO-bound, CPU-bound, or single-threaded tasks, ensuring optimal performance across different computation types.