If you are tired of writing the same messy threading or asyncio code just to run a function in the background, here is my minimalist solution.

Github: https://github.com/gunakkoc/async_obj

What My Project Does

async_obj allows running any function asynchronously. It creates a class that pretends to be whatever object/function that is passed to it and intercepts the function calls to run it in a dedicated thread. It is essentially a two-liner. Therefore, async_obj enables async operations while minimizing the code-bloat, requiring no changes in the code structure, and consuming nearly no extra resources.

Features:

Collect results of the function In case of exceptions, it is properly raised and only when result is being collected. Can check for completion OR wait/block until completion. Auto-complete works on some IDEs

Target Audience

I am using this to orchestrate several devices in a robotics setup. I believe it can be useful for anyone who deals with blocking functions such as:

Digital laboratory developers Database users Web developers Data scientist dealing with large data or computationally intense functions When quick prototyping of async operations is desired

Comparison

One can always use multithreading library. At minimum it will require wrapping the function inside another function to get the returned result. Handling errors is less controllable. Same with ThreadPoolExecutor. Multiprocessing is only worth the hassle if the aim is to distribute a computationally expensive task (i.e., running on multiple cores). Asyncio is more comprehensive but requires a lot of modification to the code with different keywords/decorators. I personally find it not so elegant.

Usage Examples

Here are some minimal examples:

from time import sleep from async_obj import async_obj class my_obj(): #a dummy class for demo def __init__(self): pass def some_func(self, val): sleep(3) # Simulate some long function return val*val x = my_obj() async_x = async_obj(x) #create a virtual async version of the object x async_x.some_func(2) # Run the original function but through the async_obj while True: done = async_x.async_obj_is_done() # Check if the function is done if done: break #do something else print(« Doing something else while waiting… ») sleep(1) result = async_x.async_obj_get_result() # Get the result or raise any exceptions # OR async_x.some_func(3) # Run the original function but through the async_obj result = async_x.async_obj_wait() # Block until completed, and get the result (or raise exception) # Same functionalities are also available when wrapping a function directly async_sleep = async_obj(sleep) #create an async version of the sleep function async_sleep(3)