CPU bound code running efficiently in async mode

background_awaitable.py

"""Background tasks functionality.

Reference: https://eng.lifion.com/navigating-cpu-bound-code-in-an-async-world-a9dcdbfbab5c
"""
import asyncio
from typing import Optional, Union
from functools import wraps, partial
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor


def background_awaitable(
    func=None,
    *,
    loop: Optional[asyncio.AbstractEventLoop] = None,
    executor: Optional[Union[ThreadPoolExecutor, ProcessPoolExecutor]] = None
):
    """
    Background awaitable decorator.
    Wrap any synchronous method or function with this
    decorator for non-blocking background execution.
    For fire-and-forget, simply call the function and method.
    It will execute in the background in a non-blocking fashion.
    To wait for the result (non-blocking), simply await the Future.
    You can await future objects at later points in your program to
    simply retrieve the result or handle any exceptions.
    Note: fire-and-forget behavior does not raise Exceptions on
    the main event loop. Awaiting the future does.
    """
    # use partial func to make args optional
    if func is None:
        return partial(background_awaitable, loop=loop, executor=executor)

    # get or create event loop if None
    if loop is None:
        loop = asyncio.get_event_loop()

    @wraps(func)
    def wrapper(*args, **kwargs) -> asyncio.Future:
        # use non-local loop
        nonlocal loop
        # self heal loop on close
        if loop.is_closed():
            loop = asyncio.get_event_loop()
        # bind kwargs and run in passed executor ->
        # if None, will run in asyncio loop's default executor
        wrapped_func = partial(func, **kwargs)
        awaitable = loop.run_in_executor(
            executor, wrapped_func, *args
        )
        return awaitable

    return wrapper