Run async code in sync (helpful for piecemeal conversion to async within codebase)

run_sync.py

from concurrent.futures import ThreadPoolExecutor
import asyncio


def run_sync(task):
    try:
        loop = asyncio.get_event_loop()
    except RuntimeError:
        loop = None
    if not loop or loop.is_running():
        with ThreadPoolExecutor(1) as pool:
            return pool.submit(asyncio.run, task).result()
    else:
        return loop.run_until_complete(task)

if __name__ == "__main__":

    async def func1(num):
        return num + 1

    async def func2(num):
        return run_sync(func1(num+1))

    def func3(num):
        return run_sync(func2(num+1))

    async def func4(num):
        return func3(num+1)

    def main(num):
        return run_sync(func4(num+1))

    assert main(0) == 5

run_worker.py

from concurrent.futures import ThreadPoolExecutor
import asyncio
import threading
import atexit


_worker_thread = None
_worker_loop = None


def shutdown():
    def close():
        if any(
            task
            for task in asyncio.Task.all_tasks()
            if task != asyncio.Task.current_task() and not task.done()
        ):
            _worker_loop.call_soon(close)
        else:
            _worker_loop.stop()

    _worker_loop.call_soon_threadsafe(close)
    _worker_thread.join()


def run_in_worker(task):
    global _worker_thread, _worker_loop
    if not _worker_thread:
        # Build base worker
        def bootstrap(loop):
            asyncio.set_event_loop(loop)
            try:
                loop.run_forever()
            finally:
                loop.close()
        _worker_loop = asyncio.new_event_loop()
        _worker_thread = threading.Thread(target=bootstrap, args=(_worker_loop,))
        _worker_thread.setDaemon(True)
        _worker_thread.start()
        atexit.register(shutdown)
    if threading.current_thread() == threading.main_thread():
        return asyncio.run_coroutine_threadsafe(task, _worker_loop)
    with ThreadPoolExecutor(1) as pool:
        return pool.submit(asyncio.run, task)



if __name__ == "__main__":

    async def func1(num):
        return num + 1

    async def func2(num):
        return run_in_worker(func1(num+1)).result()

    def func3(num):
        return run_in_worker(func2(num+1)).result()

    async def func4(num):
        return func3(num+1)

    def main(num):
        return run_in_worker(func4(num+1)).result()

    print(">>", main(0))
    assert main(0) == 5

    async def test():
        print("HERE")
        await asyncio.sleep(1)
        print("STILL HERE")

    run_in_worker(test())