aiotools
========
[](https://pypi.org/project/aiotools/)
[](https://codecov.io/gh/achimnol/aiotools)
Idiomatic asyncio utilties
*NOTE:* This project is under early stage of development. The public APIs may break version by version.
Modules
-------
* [Async Context Manager](http://aiotools.readthedocs.io/en/latest/aiotools.context.html)
* [Async Fork](http://aiotools.readthedocs.io/en/latest/aiotools.fork.html)
* [Async Functools](http://aiotools.readthedocs.io/en/latest/aiotools.func.html)
* [Async Itertools](http://aiotools.readthedocs.io/en/latest/aiotools.iter.html)
* [Async Server](http://aiotools.readthedocs.io/en/latest/aiotools.server.html)
* [Async TaskGroup](http://aiotools.readthedocs.io/en/latest/aiotools.taskgroup.html)
* [Async Timer](http://aiotools.readthedocs.io/en/latest/aiotools.timer.html)
I also recommend to try the following asyncio libraries for your happier life.
* [async_timeout](https://github.com/aio-libs/async-timeout): Provides a light-weight timeout wrapper that does not spawn subtasks.
* [aiojobs](https://github.com/aio-libs/aiojobs): Provides a concurrency-limited scheduler for asyncio tasks with graceful shutdown.
* [trio](https://github.com/python-trio/trio): An alternative implementation of asynchronous IO stack for Python, with focus on cancellation scopes and task groups called "nursery".
Examples
--------
### Async Context Manager
This is an asynchronous version of `contextlib.contextmanager` to make it
easier to write asynchronous context managers without creating boilerplate
classes.
```python
import asyncio
import aiotools
@aiotools.actxmgr
async def mygen(a):
await asyncio.sleep(1)
yield a + 1
await asyncio.sleep(1)
async def somewhere():
async with mygen(1) as b:
assert b == 2
```
Note that you need to wrap `yield` with a try-finally block to
ensure resource releases (e.g., locks), even in the case when
an exception is ocurred inside the async-with block.
```python
import asyncio
import aiotools
lock = asyncio.Lock()
@aiotools.actxmgr
async def mygen(a):
await lock.acquire()
try:
yield a + 1
finally:
lock.release()
async def somewhere():
try:
async with mygen(1) as b:
raise RuntimeError('oops')
except RuntimeError:
print('caught!') # you can catch exceptions here.
```
You can also create a group of async context managers, which
are entered/exited all at once using `asyncio.gather()`.
```python
import asyncio
import aiotools
@aiotools.actxmgr
async def mygen(a):
yield a + 10
async def somewhere():
ctxgrp = aiotools.actxgroup(mygen(i) for i in range(10))
async with ctxgrp as values:
assert len(values) == 10
for i in range(10):
assert values[i] == i + 10
```
### Async Server
This implements a common pattern to launch asyncio-based server daemons.
```python
import asyncio
import aiotools
async def echo(reader, writer):
data = await reader.read(100)
writer.write(data)
await writer.drain()
writer.close()
@aiotools.server
async def myworker(loop, pidx, args):
server = await asyncio.start_server(echo, '0.0.0.0', 8888,
reuse_port=True, loop=loop)
print(f'[{pidx}] started')
yield # wait until terminated
server.close()
await server.wait_closed()
print(f'[{pidx}] terminated')
if __name__ == '__main__':
# Run the above server using 4 worker processes.
aiotools.start_server(myworker, num_workers=4)
```
It handles SIGINT/SIGTERM signals automatically to stop the server,
as well as lifecycle management of event loops running on multiple processes.
Internally it uses `aiotools.fork` module to get kernel support to resolve
potential signal/PID related races via PID file descriptors on supported versions
(Python 3.9+ and Linux kernel 5.4+).
### Async TaskGroup
A `TaskGroup` object manages the lifecycle of sub-tasks spawned via its `create_task()`
method by guarding them with an async context manager which exits only when all sub-tasks
are either completed or cancelled.
This is motivated from [trio's nursery API](https://trio.readthedocs.io/en/stable/reference-core.html#nurseries-and-spawning)
and a draft implementation is adopted from [EdgeDB's Python client library](https://github.com/edgedb/edgedb-python).
```python
import aiotools
async def do():
async with aiotools.TaskGroup() as tg:
tg.create_task(...)
tg.create_task(...)
...
# at this point, all subtasks are either cancelled or done.
```
### Async Timer
```python
import aiotools
i = 0
async def mytick(interval):
print(i)
i += 1
async def somewhere():
t = aiotools.create_timer(mytick, 1.0)
...
t.cancel()
await t
```
`t` is an `asyncio.Task` object.
To stop the timer, call `t.cancel(); await t`.
Please don't forget `await`-ing `t` because it requires extra steps to
cancel and await all pending tasks.
To make your timer function to be cancellable, add a try-except clause
catching `asyncio.CancelledError` since we use it as a termination
signal.
You may add `TimerDelayPolicy` argument to control the behavior when the
timer-fired task takes longer than the timer interval.
`DEFAULT` is to accumulate them and cancel all the remainings at once when
the timer is cancelled.
`CANCEL` is to cancel any pending previously fired tasks on every interval.
```python
import asyncio
import aiotools
async def mytick(interval):
await asyncio.sleep(100) # cancelled on every next interval.
async def somewhere():
t = aiotools.create_timer(mytick, 1.0, aiotools.TimerDelayPolicy.CANCEL)
...
t.cancel()
await t
```
#### Virtual Clock
It provides a virtual clock that advances the event loop time instantly upon
any combination of `asyncio.sleep()` calls in multiple coroutine tasks,
by temporarily patching the event loop selector.
This is also used in [our timer test suite](https://github.com/achimnol/aiotools/blob/master/tests/test_timer.py).
```python
import aiotools
import pytest
@pytest.mark.asyncio
async def test_sleeps():
loop = aiotools.compat.get_running_loop()
vclock = aiotools.VirtualClock()
with vclock.patch_loop():
print(loop.time()) # -> prints 0
await asyncio.sleep(3600)
print(loop.time()) # -> prints 3600
```
Raw data
{
"_id": null,
"home_page": "https://github.com/achimnol/aiotools",
"name": "async-box2",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8, <3.11",
"maintainer_email": "",
"keywords": "",
"author": "Joongi Kim",
"author_email": "me@daybreaker.info",
"download_url": "https://files.pythonhosted.org/packages/8e/b6/33a178771944369044e5a3c6d69f6ae36fe094b6d75b9375c781953eab35/async-box2-1.4.9.tar.gz",
"platform": null,
"description": "aiotools\r\n========\r\n\r\n[](https://pypi.org/project/aiotools/)\r\n\r\n[](https://codecov.io/gh/achimnol/aiotools)\r\n\r\nIdiomatic asyncio utilties\r\n\r\n*NOTE:* This project is under early stage of development. The public APIs may break version by version.\r\n\r\n\r\nModules\r\n-------\r\n\r\n* [Async Context Manager](http://aiotools.readthedocs.io/en/latest/aiotools.context.html)\r\n* [Async Fork](http://aiotools.readthedocs.io/en/latest/aiotools.fork.html)\r\n* [Async Functools](http://aiotools.readthedocs.io/en/latest/aiotools.func.html)\r\n* [Async Itertools](http://aiotools.readthedocs.io/en/latest/aiotools.iter.html)\r\n* [Async Server](http://aiotools.readthedocs.io/en/latest/aiotools.server.html)\r\n* [Async TaskGroup](http://aiotools.readthedocs.io/en/latest/aiotools.taskgroup.html)\r\n* [Async Timer](http://aiotools.readthedocs.io/en/latest/aiotools.timer.html)\r\n\r\nI also recommend to try the following asyncio libraries for your happier life.\r\n\r\n* [async_timeout](https://github.com/aio-libs/async-timeout): Provides a light-weight timeout wrapper that does not spawn subtasks.\r\n* [aiojobs](https://github.com/aio-libs/aiojobs): Provides a concurrency-limited scheduler for asyncio tasks with graceful shutdown.\r\n* [trio](https://github.com/python-trio/trio): An alternative implementation of asynchronous IO stack for Python, with focus on cancellation scopes and task groups called \"nursery\".\r\n\r\n\r\nExamples\r\n--------\r\n\r\n### Async Context Manager\r\n\r\nThis is an asynchronous version of `contextlib.contextmanager` to make it\r\neasier to write asynchronous context managers without creating boilerplate\r\nclasses.\r\n\r\n```python\r\nimport asyncio\r\nimport aiotools\r\n\r\n@aiotools.actxmgr\r\nasync def mygen(a):\r\n await asyncio.sleep(1)\r\n yield a + 1\r\n await asyncio.sleep(1)\r\n\r\nasync def somewhere():\r\n async with mygen(1) as b:\r\n assert b == 2\r\n```\r\n\r\nNote that you need to wrap `yield` with a try-finally block to\r\nensure resource releases (e.g., locks), even in the case when\r\nan exception is ocurred inside the async-with block.\r\n\r\n```python\r\nimport asyncio\r\nimport aiotools\r\n\r\nlock = asyncio.Lock()\r\n\r\n@aiotools.actxmgr\r\nasync def mygen(a):\r\n await lock.acquire()\r\n try:\r\n yield a + 1\r\n finally:\r\n lock.release()\r\n\r\nasync def somewhere():\r\n try:\r\n async with mygen(1) as b:\r\n raise RuntimeError('oops')\r\n except RuntimeError:\r\n print('caught!') # you can catch exceptions here.\r\n```\r\n\r\nYou can also create a group of async context managers, which\r\nare entered/exited all at once using `asyncio.gather()`.\r\n\r\n```python\r\nimport asyncio\r\nimport aiotools\r\n\r\n@aiotools.actxmgr\r\nasync def mygen(a):\r\n yield a + 10\r\n\r\nasync def somewhere():\r\n ctxgrp = aiotools.actxgroup(mygen(i) for i in range(10))\r\n async with ctxgrp as values:\r\n assert len(values) == 10\r\n for i in range(10):\r\n assert values[i] == i + 10\r\n```\r\n\r\n### Async Server\r\n\r\nThis implements a common pattern to launch asyncio-based server daemons.\r\n\r\n```python\r\nimport asyncio\r\nimport aiotools\r\n\r\nasync def echo(reader, writer):\r\n data = await reader.read(100)\r\n writer.write(data)\r\n await writer.drain()\r\n writer.close()\r\n\r\n@aiotools.server\r\nasync def myworker(loop, pidx, args):\r\n server = await asyncio.start_server(echo, '0.0.0.0', 8888,\r\n reuse_port=True, loop=loop)\r\n print(f'[{pidx}] started')\r\n yield # wait until terminated\r\n server.close()\r\n await server.wait_closed()\r\n print(f'[{pidx}] terminated')\r\n\r\nif __name__ == '__main__':\r\n # Run the above server using 4 worker processes.\r\n aiotools.start_server(myworker, num_workers=4)\r\n```\r\n\r\nIt handles SIGINT/SIGTERM signals automatically to stop the server,\r\nas well as lifecycle management of event loops running on multiple processes.\r\nInternally it uses `aiotools.fork` module to get kernel support to resolve\r\npotential signal/PID related races via PID file descriptors on supported versions\r\n(Python 3.9+ and Linux kernel 5.4+).\r\n\r\n\r\n### Async TaskGroup\r\n\r\nA `TaskGroup` object manages the lifecycle of sub-tasks spawned via its `create_task()`\r\nmethod by guarding them with an async context manager which exits only when all sub-tasks\r\nare either completed or cancelled.\r\n\r\nThis is motivated from [trio's nursery API](https://trio.readthedocs.io/en/stable/reference-core.html#nurseries-and-spawning)\r\nand a draft implementation is adopted from [EdgeDB's Python client library](https://github.com/edgedb/edgedb-python).\r\n\r\n```python\r\nimport aiotools\r\n\r\nasync def do():\r\n async with aiotools.TaskGroup() as tg:\r\n tg.create_task(...)\r\n tg.create_task(...)\r\n ...\r\n # at this point, all subtasks are either cancelled or done.\r\n```\r\n\r\n\r\n### Async Timer\r\n\r\n```python\r\nimport aiotools\r\n\r\ni = 0\r\n\r\nasync def mytick(interval):\r\n print(i)\r\n i += 1\r\n\r\nasync def somewhere():\r\n t = aiotools.create_timer(mytick, 1.0)\r\n ...\r\n t.cancel()\r\n await t\r\n```\r\n\r\n`t` is an `asyncio.Task` object.\r\nTo stop the timer, call `t.cancel(); await t`.\r\nPlease don't forget `await`-ing `t` because it requires extra steps to\r\ncancel and await all pending tasks.\r\nTo make your timer function to be cancellable, add a try-except clause\r\ncatching `asyncio.CancelledError` since we use it as a termination\r\nsignal.\r\n\r\nYou may add `TimerDelayPolicy` argument to control the behavior when the\r\ntimer-fired task takes longer than the timer interval.\r\n`DEFAULT` is to accumulate them and cancel all the remainings at once when\r\nthe timer is cancelled.\r\n`CANCEL` is to cancel any pending previously fired tasks on every interval.\r\n\r\n```python\r\nimport asyncio\r\nimport aiotools\r\n\r\nasync def mytick(interval):\r\n await asyncio.sleep(100) # cancelled on every next interval.\r\n\r\nasync def somewhere():\r\n t = aiotools.create_timer(mytick, 1.0, aiotools.TimerDelayPolicy.CANCEL)\r\n ...\r\n t.cancel()\r\n await t\r\n```\r\n\r\n#### Virtual Clock\r\n\r\nIt provides a virtual clock that advances the event loop time instantly upon\r\nany combination of `asyncio.sleep()` calls in multiple coroutine tasks,\r\nby temporarily patching the event loop selector.\r\n\r\nThis is also used in [our timer test suite](https://github.com/achimnol/aiotools/blob/master/tests/test_timer.py).\r\n\r\n```python\r\nimport aiotools\r\nimport pytest\r\n\r\n@pytest.mark.asyncio\r\nasync def test_sleeps():\r\n loop = aiotools.compat.get_running_loop()\r\n vclock = aiotools.VirtualClock()\r\n with vclock.patch_loop():\r\n print(loop.time()) # -> prints 0\r\n await asyncio.sleep(3600)\r\n print(loop.time()) # -> prints 3600\r\n```\r\n",
"bugtrack_url": null,
"license": "apache2",
"summary": "Idiomatic asyncio utilities",
"version": "1.4.9",
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "8eb633a178771944369044e5a3c6d69f6ae36fe094b6d75b9375c781953eab35",
"md5": "b84421fb2668432260434f402377a51f",
"sha256": "72c39a64174f510d081b8127eedbc6bc0ce9375930fad78327d7a12f05795dc0"
},
"downloads": -1,
"filename": "async-box2-1.4.9.tar.gz",
"has_sig": false,
"md5_digest": "b84421fb2668432260434f402377a51f",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8, <3.11",
"size": 25823,
"upload_time": "2023-04-04T19:01:57",
"upload_time_iso_8601": "2023-04-04T19:01:57.532688Z",
"url": "https://files.pythonhosted.org/packages/8e/b6/33a178771944369044e5a3c6d69f6ae36fe094b6d75b9375c781953eab35/async-box2-1.4.9.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-04-04 19:01:57",
"github": true,
"gitlab": false,
"bitbucket": false,
"github_user": "achimnol",
"github_project": "aiotools",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "async-box2"
}
JSON
