| Name | aioshutdown JSON |
| Version |
0.0.2
JSON |
| download |
| home_page | |
| Summary | Context manager that provides simple graceful shutdown interface for your asyncio tasks. |
| upload_time | 2023-03-13 20:13:29 |
| maintainer | |
| docs_url | None |
| author | |
| requires_python | >=3.8 |
| license | |
| keywords |
asyncio
graceful
shutdown
|
| VCS |
|
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# aioshutdown
Context manager that provides simple graceful shutdown interface for your asyncio tasks.
## Installation
```
pip install -U aioshutdown
```
## Usage
Just specify the list of signals that you want to intercept using the `|` operator:
```python
from aioshutdown import SIGTERM, SIGINT, SIGHUP
with SIGTERM | SIGHUP | SIGINT as loop:
...
```
The context manager will return an instance of the current event loop. You can intercept signals directly inside your coroutines:
```python
async def my_task(sleep: int):
try:
""" Main logic of your coroutine. """
except asyncio.CancelledError as exc:
""" In this place you can gracefully complete the work. """
```
Full example:
```python
import asyncio
from aioshutdown import SIGTERM, SIGINT, SIGHUP
import logging
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s,%(msecs)d %(levelname)s: %(message)s",
datefmt="%H:%M:%S",
)
async def my_task(sleep: int):
try:
while True:
logging.info("Sleep from task #%s", sleep)
await asyncio.sleep(sleep)
except asyncio.CancelledError as exc:
# Received an exit signal. In this place we gracefully complete the work.
signal = exc.args[0] # NOTE: works in Python >= 3.9 only: https://docs.python.org/3/library/asyncio-future.html?highlight=Changed%20in%20version%203.9:%20Added%20the%20msg%20parameter.#asyncio.Future.cancel
logging.warning("Received %s signal. Shutting down...", signal.value)
# Usage with `run_forever`
with SIGTERM | SIGHUP | SIGINT as loop:
tasks = [loop.create_task(my_task(i)) for i in range(1, 10)]
loop.run_forever()
# Usage with `run_until_complete`
with SIGTERM | SIGHUP | SIGINT as loop:
tasks = [loop.create_task(my_task(i)) for i in range(1, 10)]
loop.run_until_complete(asyncio.gather(*results))
```
Output
```
23:04:05,798 INFO: Sleep from task #1
23:04:05,798 INFO: Sleep from task #2
23:04:06,799 INFO: Sleep from task #1
23:04:07,800 INFO: Sleep from task #2
23:04:07,800 INFO: Sleep from task #1
23:04:08,801 INFO: Sleep from task #1
^C23:04:09,504 INFO: Received exit signal SIGINT...
23:04:09,504 INFO: Cancelling 2 outstanding tasks
23:04:09,504 WARNING: Received 2 signal. Shutting down...
23:04:09,504 WARNING: Received 2 signal. Shutting down...
23:04:09,504 INFO: Stopping event loop
```
## Useful links
[Graceful Shutdowns with asyncio](https://www.roguelynn.com/words/asyncio-graceful-shutdowns/)
Raw data
{
"_id": null,
"home_page": "",
"name": "aioshutdown",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": "",
"keywords": "asyncio,graceful,shutdown",
"author": "",
"author_email": "Ali Aliyev <ali@aliev.me>",
"download_url": "https://files.pythonhosted.org/packages/c9/85/b6cd93a9d60b02547124a93090f5163bde9bdc6b1b56e58c218af892a62f/aioshutdown-0.0.2.tar.gz",
"platform": null,
"description": "# aioshutdown\n\nContext manager that provides simple graceful shutdown interface for your asyncio tasks.\n\n## Installation\n\n```\npip install -U aioshutdown\n```\n\n## Usage\n\nJust specify the list of signals that you want to intercept using the `|` operator:\n\n```python\nfrom aioshutdown import SIGTERM, SIGINT, SIGHUP\n\n\nwith SIGTERM | SIGHUP | SIGINT as loop:\n ...\n```\n\nThe context manager will return an instance of the current event loop. You can intercept signals directly inside your coroutines:\n\n```python\nasync def my_task(sleep: int):\n try:\n \"\"\" Main logic of your coroutine. \"\"\"\n except asyncio.CancelledError as exc:\n \"\"\" In this place you can gracefully complete the work. \"\"\"\n```\n\nFull example:\n\n\n```python\nimport asyncio\nfrom aioshutdown import SIGTERM, SIGINT, SIGHUP\nimport logging\n\n\nlogging.basicConfig(\n level=logging.INFO,\n format=\"%(asctime)s,%(msecs)d %(levelname)s: %(message)s\",\n datefmt=\"%H:%M:%S\",\n)\n\nasync def my_task(sleep: int):\n try:\n while True:\n logging.info(\"Sleep from task #%s\", sleep)\n await asyncio.sleep(sleep)\n except asyncio.CancelledError as exc:\n # Received an exit signal. In this place we gracefully complete the work.\n signal = exc.args[0] # NOTE: works in Python >= 3.9 only: https://docs.python.org/3/library/asyncio-future.html?highlight=Changed%20in%20version%203.9:%20Added%20the%20msg%20parameter.#asyncio.Future.cancel\n logging.warning(\"Received %s signal. Shutting down...\", signal.value)\n\n# Usage with `run_forever`\n\nwith SIGTERM | SIGHUP | SIGINT as loop:\n tasks = [loop.create_task(my_task(i)) for i in range(1, 10)]\n loop.run_forever()\n\n# Usage with `run_until_complete`\n\nwith SIGTERM | SIGHUP | SIGINT as loop:\n tasks = [loop.create_task(my_task(i)) for i in range(1, 10)]\n loop.run_until_complete(asyncio.gather(*results))\n```\n\nOutput\n\n```\n23:04:05,798 INFO: Sleep from task #1\n23:04:05,798 INFO: Sleep from task #2\n23:04:06,799 INFO: Sleep from task #1\n23:04:07,800 INFO: Sleep from task #2\n23:04:07,800 INFO: Sleep from task #1\n23:04:08,801 INFO: Sleep from task #1\n^C23:04:09,504 INFO: Received exit signal SIGINT...\n23:04:09,504 INFO: Cancelling 2 outstanding tasks\n23:04:09,504 WARNING: Received 2 signal. Shutting down...\n23:04:09,504 WARNING: Received 2 signal. Shutting down...\n23:04:09,504 INFO: Stopping event loop\n```\n\n\n## Useful links\n\n[Graceful Shutdowns with asyncio](https://www.roguelynn.com/words/asyncio-graceful-shutdowns/)\n",
"bugtrack_url": null,
"license": "",
"summary": "Context manager that provides simple graceful shutdown interface for your asyncio tasks.",
"version": "0.0.2",
"split_keywords": [
"asyncio",
"graceful",
"shutdown"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "015a7a89cb1a994e962be9d88567e81d92b34d8ce7d227709e1290bd36badf7e",
"md5": "b531affc22b2ff8265d676e0dd780763",
"sha256": "ed75f50ff5a670a4c5cb12b1ecfe5ceb5298e8a2f13de32d52c92d61f4c07a3f"
},
"downloads": -1,
"filename": "aioshutdown-0.0.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "b531affc22b2ff8265d676e0dd780763",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 5080,
"upload_time": "2023-03-13T20:13:28",
"upload_time_iso_8601": "2023-03-13T20:13:28.062784Z",
"url": "https://files.pythonhosted.org/packages/01/5a/7a89cb1a994e962be9d88567e81d92b34d8ce7d227709e1290bd36badf7e/aioshutdown-0.0.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "c985b6cd93a9d60b02547124a93090f5163bde9bdc6b1b56e58c218af892a62f",
"md5": "1ad54ecde2757c3464a56bfac8c5dff0",
"sha256": "2a69baf10f4235f51eb3228868406abe39b345081c5c1930e8ce17d4e5a7597d"
},
"downloads": -1,
"filename": "aioshutdown-0.0.2.tar.gz",
"has_sig": false,
"md5_digest": "1ad54ecde2757c3464a56bfac8c5dff0",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 4858,
"upload_time": "2023-03-13T20:13:29",
"upload_time_iso_8601": "2023-03-13T20:13:29.585058Z",
"url": "https://files.pythonhosted.org/packages/c9/85/b6cd93a9d60b02547124a93090f5163bde9bdc6b1b56e58c218af892a62f/aioshutdown-0.0.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2023-03-13 20:13:29",
"github": false,
"gitlab": false,
"bitbucket": false,
"lcname": "aioshutdown"
}
