async wrapper
Project description
async-wrapper
how to install
$ pip install async_wrapper
# or
$ pip install "async_wrapper[all]"
# or
$ pip install "async_wrapper[loky]"
how to use
from __future__ import annotations
import time
import anyio
from async_wrapper import TaskGroupWrapper, async_to_sync
@async_to_sync("thread")
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.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
async_wrapper-0.3.2.tar.gz
(7.6 kB
view hashes)
Built Distribution
Close
Hashes for async_wrapper-0.3.2-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 82f54968aec0052f729107a77573c9417655a1b57f11ef82727c34a4e01c5feb |
|
MD5 | 7d90bb84940b83c79d873843e80b2acd |
|
BLAKE2b-256 | 3672055124a5e2cca3db56c2e4806b3ad741a67d44ce5d9f7d392498fec6c2bd |