Asyncio/Curio/Trio compatibility module

compat.py

import asyncio

try:
    import curio
except ImportError:
    curio = None

try:
    import trio
except ImportError:
    trio = None


async def get_async_module():
    try:
        if trio and trio.hazmat.current_task():
            try:
                if asyncio and asyncio.get_event_loop():
                    return asyncio
            except RuntimeError:
                pass
            return trio
    except RuntimeError:
        pass
    try:
        if curio and (await curio.current_task()):
            return curio
    except RuntimeError:
        pass
    try:
        if asyncio.get_event_loop().is_running():
            return asyncio
    except RuntimeError:
        pass
    raise RuntimeError(
        'The current asynchronous framework is not supported')

    
# Framework agnostic sleep

async def sleep(seconds):
    module = await get_async_module()
    return await module.sleep(seconds)


# Testing

def test_trio():
    trio.run(sleep, 1)


def test_curio():
    curio.run(sleep(1))


def test_asyncio():
    loop = asyncio.get_event_loop()
    loop.run_until_complete(sleep(1))