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