async-wrapper

how to install

$ pip install async_wrapper

how to use

from __future__ import annotations

import time

import anyio

from async_wrapper import TaskGroupWrapper, toggle_func


@toggle_func
async def sample_func() -> int:
    await anyio.sleep(1)
    return 1


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


def main():
    result = sample_func()
    assert isinstance(result, int)
    assert result == 1


async def async_main():
    semaphore = anyio.Semaphore(2)

    start = time.perf_counter()
    async with anyio.create_task_group() as task_group:
        wrapper = TaskGroupWrapper(task_group)
        func = wrapper.wrap(sample_func_2, semaphore)
        value_1 = func(1)
        value_2 = func(2)
        value_3 = func(3)
    end = time.perf_counter()

    assert isinstance(value_1.value, int)
    assert isinstance(value_2.value, int)
    assert isinstance(value_3.value, int)
    assert value_1.value == 1
    assert value_2.value == 2
    assert value_3.value == 3
    assert 1.5 < end - start < 2.5


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

License

MIT, see LICENSE.