Reuse existing async methods

async_to_sync.py

# from: https://gist.github.com/phizaz/20c36c6734878c6ec053245a477572ec
import asyncio
import functools
import asyncio

def force_async(fn):
    '''
    turns a sync function to async function using threads
    '''
    from concurrent.futures import ThreadPoolExecutor
    import asyncio
    pool = ThreadPoolExecutor()

    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        future = pool.submit(fn, *args, **kwargs)
        return asyncio.wrap_future(future)  # make it awaitable

    return wrapper


def force_sync(fn):
    '''
    turn an async function to sync function
    '''
    import asyncio

    @functools.wraps(fn)
    def wrapper(*args, **kwargs):
        res = fn(*args, **kwargs)
        if asyncio.iscoroutine(res):
            return asyncio.get_event_loop().run_until_complete(res)
        return res

    return wrapper



###### USAGE ######

class ClientAsync:

    def __init__(self): 
        pass

    async def take_a_nap(self, delay=1):
        import asyncio
        await asyncio.sleep(delay)
        return f"Slept for {delay} seconds"


class ClientSync:

    def __init__(self):
        self.async_client = ClientAsync()

    @force_sync
    async def take_a_nap(self, delay=1):
        return await self.async_client.take_a_nap(delay)



# Sync usage
my_sync_client = ClientSync()
result_sync = my_sync_client.take_a_nap(0.5)
print(f"result_sync = {result_sync}")


# Async usage
my_async_client = ClientAsync()
event_loop = asyncio.get_event_loop()
result_async = event_loop.run_until_complete(
    my_async_client.take_a_nap(0.75)
)
print(f"result_async = {result_async}")