async-wrapper 0.1.3

async wrapper

async-wrapper

how to install

$ pip install async_wrapper
# or
$ pip install "async_wrapper[all]"
# or
$ pip install "async_wrapper[anyio]"
# or
$ pip install "async_wrapper[loky]"

how to use

from __future__ import annotations

import asyncio

# or(>=py311)
# from asyncio.taskgroups import TaskGroup
from async_wrapper import async_to_sync, get_task_group_factory, get_task_group_wrapper


@async_to_sync("thread")
async def sample_func() -> int:
    await asyncio.sleep(1)
    return 1


result = sample_func()
assert isinstance(result, int)
assert result == 1


async def sample_func_2(x: int) -> int:
    await asyncio.sleep(1)
    return x


async def main():
    wrapper = get_task_group_wrapper("asyncio")
    factory = get_task_group_factory("asyncio")
    async with factory() as task_group:
        value_1 = wrapper(sample_func_2, task_group)(1)
        value_2 = wrapper(sample_func_2, task_group)(2)

    assert isinstance(value_1.value, int)
    assert isinstance(value_2.value, int)
    assert value_1.value == 1
    assert value_2.value == 2

License

MIT, see LICENSE.