| Name | async-timer JSON |
| Version |
1.1.6
JSON |
| download |
| home_page | |
| Summary | The missing Python async timer. |
| upload_time | 2024-01-24 23:04:25 |
| maintainer | |
| docs_url | None |
| author | Ilya O. |
| requires_python | >=3.8,<4.0 |
| license | MIT |
| keywords |
async
timer
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Async timer
This package provides an async timer object, that should have been part of batteries.
[](docs/badges/tests.svg)
[](docs/badges/coverage.svg)
[](https://github.com/VRGhost/async-timer/actions/workflows/main.yml)
## Purpose
Sometimes, you need a way to make something happen over and over again at certain times, like updating information or sending reminders. That's where Async Timer comes in. It lets you set up these repeated actions easily.
This package is particularly useful for tasks like automatically updating caches in the background without disrupting the primary application's workflow.
## Features
* **Zero Dependencies**: Written entirely in Python, Async Timer operates independently without needing any external libraries.
* **Versatility in Callables**: It accommodates various callable types, including:
* Synchronous functions
* Asynchronous functions
* Synchronous generators
* Asynchronous generators
* **Wait for the Next Tick**: You can set it up so your program waits for the timer to do its thing, and then continues.
* **Keep Getting Updates**: You can use it in a loop to keep getting updates every time the timer goes off.
* **Cancel anytime**: The timer object can be stopped at any time either explicitly by calling `stop()`/`cancel()` method OR it can stop automatically on an awaitable resolving (the `cancel_aws` constructor artument)
* **Test friendly**: The package provides an additional `mock_async_timer.MockTimer` class with mocked sleep function to aid in your testing
## Example Usage
### FastAPI
This snippet starts fastapi webserver with the `refresh_db` function being executed every 5 seconds, refresing a shared `DB_CACHE` object.
```python
import contextlib
import time
import uvicorn
from fastapi import FastAPI
import async_timer
DB_CACHE = {"initialised": False}
async def refresh_db():
global DB_CACHE
DB_CACHE |= {"initialised": True, "cur_value": time.time()}
@contextlib.asynccontextmanager
async def lifespan(_app: FastAPI):
async with async_timer.Timer(delay=5, target=refresh_db) as timer:
await timer.wait(hit_count=1) # block until the timer triggers at least once
yield
app = FastAPI(lifespan=lifespan)
@app.get("/")
async def root():
return {"message": "Hello World", "db_cache": DB_CACHE}
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=8000)
```
### join()
```python
import async_timer
timer = async_timer.Timer(12, target=lambda: 42)
timer.start()
val = await timer.join() # `val` will be set to 42 after 12 seconds
```
### for loop
```python
# Async for loop example
import async_timer
import time
with async_timer.Timer(14, target=time.time) as timer:
async for time_rv in timer:
print(f"{time_rv=}") # Prints current time every 14 seconds
```
Raw data
{
"_id": null,
"home_page": "",
"name": "async-timer",
"maintainer": "",
"docs_url": null,
"requires_python": ">=3.8,<4.0",
"maintainer_email": "",
"keywords": "async,timer",
"author": "Ilya O.",
"author_email": "vrghost@gmail.com",
"download_url": "https://files.pythonhosted.org/packages/af/ee/fe0f95b436c64873795e89f0d13cbf0f4c88afae00b3ebba46363d11c13f/async_timer-1.1.6.tar.gz",
"platform": null,
"description": "# Async timer\n\nThis package provides an async timer object, that should have been part of batteries.\n\n[](docs/badges/tests.svg)\n[](docs/badges/coverage.svg)\n\n[](https://github.com/VRGhost/async-timer/actions/workflows/main.yml)\n\n## Purpose\n\nSometimes, you need a way to make something happen over and over again at certain times, like updating information or sending reminders. That's where Async Timer comes in. It lets you set up these repeated actions easily.\n\nThis package is particularly useful for tasks like automatically updating caches in the background without disrupting the primary application's workflow.\n\n## Features\n\n* **Zero Dependencies**: Written entirely in Python, Async Timer operates independently without needing any external libraries.\n* **Versatility in Callables**: It accommodates various callable types, including:\n * Synchronous functions\n * Asynchronous functions\n * Synchronous generators\n * Asynchronous generators\n* **Wait for the Next Tick**: You can set it up so your program waits for the timer to do its thing, and then continues.\n* **Keep Getting Updates**: You can use it in a loop to keep getting updates every time the timer goes off.\n* **Cancel anytime**: The timer object can be stopped at any time either explicitly by calling `stop()`/`cancel()` method OR it can stop automatically on an awaitable resolving (the `cancel_aws` constructor artument)\n* **Test friendly**: The package provides an additional `mock_async_timer.MockTimer` class with mocked sleep function to aid in your testing\n\n## Example Usage\n\n### FastAPI\n\nThis snippet starts fastapi webserver with the `refresh_db` function being executed every 5 seconds, refresing a shared `DB_CACHE` object.\n\n```python\n\nimport contextlib\nimport time\n\nimport uvicorn\nfrom fastapi import FastAPI\n\nimport async_timer\n\nDB_CACHE = {\"initialised\": False}\n\n\nasync def refresh_db():\n global DB_CACHE\n DB_CACHE |= {\"initialised\": True, \"cur_value\": time.time()}\n\n\n@contextlib.asynccontextmanager\nasync def lifespan(_app: FastAPI):\n async with async_timer.Timer(delay=5, target=refresh_db) as timer:\n await timer.wait(hit_count=1) # block until the timer triggers at least once\n yield\n\n\napp = FastAPI(lifespan=lifespan)\n\n\n@app.get(\"/\")\nasync def root():\n return {\"message\": \"Hello World\", \"db_cache\": DB_CACHE}\n\n\nif __name__ == \"__main__\":\n uvicorn.run(app, host=\"0.0.0.0\", port=8000)\n\n```\n\n### join()\n```python\n\nimport async_timer\n\ntimer = async_timer.Timer(12, target=lambda: 42)\ntimer.start()\nval = await timer.join() # `val` will be set to 42 after 12 seconds\n```\n\n### for loop\n```python\n# Async for loop example\nimport async_timer\nimport time\nwith async_timer.Timer(14, target=time.time) as timer:\n async for time_rv in timer:\n print(f\"{time_rv=}\") # Prints current time every 14 seconds\n\n```",
"bugtrack_url": null,
"license": "MIT",
"summary": "The missing Python async timer.",
"version": "1.1.6",
"project_urls": {
"Source": "https://github.com/VRGhost/async-timer"
},
"split_keywords": [
"async",
"timer"
],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "08eb21b78921bb57c690fbabc42ff3f7220f6f7310219b83311f5ebbe4462a8a",
"md5": "da61fe56390b952b0b0e47d476657940",
"sha256": "b178e0ecbcf3c5c9711d650c90a04b8691d09bb2145c123c053d382927b7e2b8"
},
"downloads": -1,
"filename": "async_timer-1.1.6-py3-none-any.whl",
"has_sig": false,
"md5_digest": "da61fe56390b952b0b0e47d476657940",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.8,<4.0",
"size": 7796,
"upload_time": "2024-01-24T23:04:23",
"upload_time_iso_8601": "2024-01-24T23:04:23.674113Z",
"url": "https://files.pythonhosted.org/packages/08/eb/21b78921bb57c690fbabc42ff3f7220f6f7310219b83311f5ebbe4462a8a/async_timer-1.1.6-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "afeefe0f95b436c64873795e89f0d13cbf0f4c88afae00b3ebba46363d11c13f",
"md5": "8e42c0803944805e28ace35f629588d5",
"sha256": "f0a07574b9ccc9ac484ec178deb4f1018150cb96f007cc12ee2a96280252c310"
},
"downloads": -1,
"filename": "async_timer-1.1.6.tar.gz",
"has_sig": false,
"md5_digest": "8e42c0803944805e28ace35f629588d5",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.8,<4.0",
"size": 7991,
"upload_time": "2024-01-24T23:04:25",
"upload_time_iso_8601": "2024-01-24T23:04:25.151699Z",
"url": "https://files.pythonhosted.org/packages/af/ee/fe0f95b436c64873795e89f0d13cbf0f4c88afae00b3ebba46363d11c13f/async_timer-1.1.6.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-01-24 23:04:25",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "VRGhost",
"github_project": "async-timer",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "async-timer"
}
