async_wrapper.convert module

async_wrapper.convert.toggle_func(func: Callable[[ParamT], ValueT] | Callable[[ParamT], Coroutine[Any, Any, ValueT]]) → Callable[[ParamT], ValueT] | Callable[[ParamT], Coroutine[Any, Any, ValueT]]

Convert between synchronous and asynchronous functions.

Parameters:: func – 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: Callable[[ParamT], ValueT_co]) → Callable[[ParamT], Coroutine[Any, Any, ValueT_co]]

Convert a synchronous function to an asynchronous function.

Parameters:: func – 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: Callable[[ParamT], Awaitable[ValueT_co]]) → Callable[[ParamT], ValueT_co]

Convert an awaitable function to a synchronous function.

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

Parameters:: func – An awaitable function.
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")
$ poetry run python main.py
asyncio <_UnixSelectorEventLoop running=True closed=False debug=False>
asyncio <_UnixSelectorEventLoop running=True closed=False debug=False>
asyncio <uvloop.Loop running=True closed=False debug=False>
trio