async_wrapper.convert module¶

async_wrapper.convert.toggle_func(func: collections.abc.Callable[_P, _T] | collections.abc.Callable[_P, Coroutine[Any, Any, _T]]) → collections.abc.Callable[_P, _T] | collections.abc.Callable[_P, Awaitable[_T]][source]¶

Convert between synchronous and asynchronous functions.

Parameters:¶

func: collections.abc.Callable[_P, _T] | collections.abc.Callable[_P, Coroutine[Any, Any, _T]]¶

A function that can be either synchronous or asynchronous.

Returns:¶

A function that matches the desired synchronicity, either synchronous or asynchronous.

async_wrapper.convert.sync_to_async(func: collections.abc.Callable[_P, _T]) → collections.abc.Callable[_P, Awaitable[_T]][source]¶

Convert a synchronous function to an asynchronous function.

Parameters:¶

func: collections.abc.Callable[_P, _T]¶

The synchronous function to be converted.

Returns:¶

An asynchronous function that behaves equivalently to the input synchronous function.

Example

import time

import anyio

from async_wrapper import sync_to_async


@sync_to_async
def test(x: int) -> int:
    print(f"[{x}] test: start")
    time.sleep(1)
    print(f"[{x}] test: end")
    return x


async def main() -> None:
    start = time.perf_counter()
    async with anyio.create_task_group() as task_group:
        for i in range(4):
            task_group.start_soon(test, i)
    end = time.perf_counter()
    assert end - start < 1.1


if __name__ == "__main__":
    anyio.run(main)

async_wrapper.convert.async_to_sync(func_or_awaitable: collections.abc.Callable[_P, Awaitable[_T]] | Awaitable[_T]) → collections.abc.Callable[_P, _T] | collections.abc.Callable[[], _T][source]¶

Convert an awaitable function or awaitable object to a synchronous function.

If used within an asynchronous context, attempts to use the same backend. Defaults to asyncio.

Parameters:¶

func_or_awaitable: collections.abc.Callable[_P, Awaitable[_T]] | Awaitable[_T]¶

An awaitable function or awaitable object.

Returns:¶

A synchronous function.

Example

import asyncio
import time

import anyio
import sniffio

from async_wrapper import async_to_sync


@async_to_sync
async def test(x: int) -> int:
    backend = sniffio.current_async_library()
    if backend == "asyncio":
        loop = asyncio.get_running_loop()
        print(backend, loop)
    else:
        print(backend)
    await anyio.sleep(1)
    return x


def main() -> None:
    start = time.perf_counter()
    result = test(1)
    end = time.perf_counter()
    assert result == 1
    assert end - start < 1.1


async def async_main() -> None:
    start = time.perf_counter()
    result = test(1)
    end = time.perf_counter()
    assert result == 1
    assert end - start < 1.1


if __name__ == "__main__":
    main()
    anyio.run(
        async_main,
        backend="asyncio",
        backend_options={"use_uvloop": True},
    )
    anyio.run(
        async_main,
        backend="asyncio",
        backend_options={"use_uvloop": True},
    )
    anyio.run(async_main, backend="trio")