| Name | aioshutdown JSON |
| Version |
0.0.4
JSON |
| download |
| home_page | None |
| Summary | Context manager that provides simple graceful shutdown interface for your asyncio tasks. |
| upload_time | 2024-12-25 15:29:42 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.8 |
| license | None |
| 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": null,
"name": "aioshutdown",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.8",
"maintainer_email": null,
"keywords": "asyncio, graceful, shutdown",
"author": null,
"author_email": "Ali Aliyev <ali@aliev.me>",
"download_url": "https://files.pythonhosted.org/packages/9a/b5/f4d1fb8063c67781eb5468c4125f2b3f651c834ca8886f03743f93b6f466/aioshutdown-0.0.4.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": null,
"summary": "Context manager that provides simple graceful shutdown interface for your asyncio tasks.",
"version": "0.0.4",
"project_urls": {
"Bug Tracker": "https://github.com/aliev/aioshutdown/issues",
"Homepage": "https://github.com/aliev/aioshutdown"
},
"split_keywords": [
"asyncio",
" graceful",
" shutdown"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "c4090480a8bb9eb22621f1f3f26004c7fab1ce40e01ff7c35493aab33361f525",
"md5": "63a5ecf3709bbe6a333b8aed2a04eb00",
"sha256": "b70d4e63f3a01184af4cf404d175c74a94cc033db623722ed0e4da908cf56338"
},
"downloads": -1,
"filename": "aioshutdown-0.0.4-py3-none-any.whl",
"has_sig": false,
"md5_digest": "63a5ecf3709bbe6a333b8aed2a04eb00",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8",
"size": 5375,
"upload_time": "2024-12-25T15:29:40",
"upload_time_iso_8601": "2024-12-25T15:29:40.695388Z",
"url": "https://files.pythonhosted.org/packages/c4/09/0480a8bb9eb22621f1f3f26004c7fab1ce40e01ff7c35493aab33361f525/aioshutdown-0.0.4-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "9ab5f4d1fb8063c67781eb5468c4125f2b3f651c834ca8886f03743f93b6f466",
"md5": "f4f20be94e498ab49fda0950fcb92cdf",
"sha256": "e0627700fbf6388d3ab4faccf7204e78eb4a1c78caa9236d5e429967df7e2f80"
},
"downloads": -1,
"filename": "aioshutdown-0.0.4.tar.gz",
"has_sig": false,
"md5_digest": "f4f20be94e498ab49fda0950fcb92cdf",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8",
"size": 5118,
"upload_time": "2024-12-25T15:29:42",
"upload_time_iso_8601": "2024-12-25T15:29:42.782594Z",
"url": "https://files.pythonhosted.org/packages/9a/b5/f4d1fb8063c67781eb5468c4125f2b3f651c834ca8886f03743f93b6f466/aioshutdown-0.0.4.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-12-25 15:29:42",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "aliev",
"github_project": "aioshutdown",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "aioshutdown"
}
