# Runevery task when it's time
Runevery (one word, as in 'bakery') is a simple async scheduling library designed to schedule tasks in _no time_. With a simple and intuitive API, you can plan and manage tasks to run at specific intervals, times, or even based on custom conditions. No more writing complex loops and time checks!
> [!NOTE]
> It is highly recommended to use type-checking with runevery. Not only did I spent some time polishing it here, but also types are cool and nice. And I won't be covering some stuff that is obvious from the names.
> [!IMPORTANT]
> This package is fresh and both the readme and the API are not very stable and finished. Feel free to create issues and stuff, but be aware that this is at its early stages.
## Installation
You can install runevery from PyPI using pip. Just run:
```sh
pip install runevery
```
Or, if you prefer Poetry:
```sh
poetry add runevery
```
For more details, check out the [PyPI page](https://pypi.org/project/runevery/).
## Basic usage
Let's start with some basic usage examples to get you familiar with runevery.
### Example: Fetching emails every 10 minutes
Suppose you want to fetch your emails every 10 minutes. Here's how you can do it:
```python
from asyncio import run as asyncio_run
from runevery import run
@run.every(minutes=10)
async def fetch_emails():
print("Very convincing code to fetch e-mails or something")
asyncio_run(run.loop())
```
### Example: Updating weather data every hour
Need to update your weather data every hour? No problem!
```python
from asyncio import run as asyncio_run
from runevery import run
@run.every(hours=1)
async def update_weather():
print("Updating weather data...")
asyncio_run(run.loop())
```
### Example: Running a task once at startup
Sometimes, you might want to run a task just once at startup. Here's how:
```python
from asyncio import run as asyncio_run
from runevery import run
@run.once()
async def startup_task():
print("Running startup task...")
asyncio_run(run.loop())
```
### Example: Running a task after a specific time
Want to run a task after a specific Unix timestamp, just for fun? We have a tool for that, it's called `scheduler.after`:
```python
from asyncio import run as asyncio_run
from runevery import run
@run.after(offset=1718567362)
async def run_after_timestamp():
print("Running after specific timestamp...")
asyncio_run(run.loop())
```
## What's a scheduler, anyway?
### What is the scheduler task loop?
The scheduler and its task loop are the core of runevery. It continuously ticks through all scheduled tasks, checking if it's time to run them based on their planners.
But how does the scheduler know when to run a task? It doesn't. **CREDITS ROLL**
Okay, but actually, it just ticks every task, and each task's planner decides if it's time to run.
### What are the arguments to scheduler methods?
Let's break down the arguments you can pass to `scheduler.plan`, `scheduler.every`, `scheduler.once`, `scheduler.at`, and `scheduler.after`.
#### Example: Using `scheduler.every`
```python
from asyncio import run as asyncio_run
from runevery import run
@run.every(seconds=30)
async def every_30_seconds():
print("Running every 30 seconds")
asyncio_run(run.loop())
```
#### Example: Using `scheduler.at`
```python
from asyncio import run as asyncio_run
from runevery import run
@run.at(interval=86400, hours=15)
async def daily_at_3pm():
print("Running daily at 3 PM")
asyncio_run(run.loop())
```
## What's a planner?
Planners are the brains behind the scheduling. They decide if a task should run based on various conditions.
### Built-in planners
#### IntervalPlanner
Runs tasks at fixed intervals.
```python
from runevery import IntervalPlanner, run
@run.plan(planner=IntervalPlanner(interval=60))
async def every_minute():
print("Running every minute")
asyncio_run(run.loop())
```
#### FixedOffsetPlanner
Runs tasks at a fixed timestamp, optionally repeating at intervals.
```python
from runevery import FixedOffsetPlanner, Scheduler
scheduler = Scheduler()
@scheduler.plan(planner=FixedOffsetPlanner(offset=1718567362, interval=3600))
async def hourly_after_timestamp():
print("Running hourly after a specific timestamp")
asyncio_run(scheduler.loop())
```
#### CooldownPlanner
Runs tasks based on a cooldown period.
```python
from runevery import CooldownPlanner, run
class MyCooldownSource:
def get_cooldown(self, interval: float):
return 0 # Replace with actual cooldown logic
@run.plan(planner=CooldownPlanner(MyCooldownSource(), interval=300))
async def cooldown_task():
print("Running with cooldown")
asyncio_run(run.loop())
```
## Interval strategy
The interval strategy defines if the interval is counted from the start or the end of the run.
- Use `"start"` (default), when you want to count interval from the start of the previous task. The duration between callback starts doesn't drift, since it doesn't depend on the task duration.
- Use `"end"`, when you care about interval _between_ tasks. This interval starts counting after the previous task has ended, so the pause between tasks doesn't drift.
```python
from runevery import run
from asyncio import sleep
@run.every(seconds=10, interval_strategy="end")
async def very_long_task():
await sleep(3600)
print("This task will run each ~3610 seconds, despite the interval beint 10 seconds")
asyncio_run(run.loop())
```
## Arguments to the task callback
Task callbacks can receive optional arguments like `task` and `scheduler`.
```python
from runevery import run, Scheduler, SchedulingTask
@run.every(minutes=5)
async def task_with_args(task: SchedulingTask, scheduler: Scheduler):
print(f"Task {task.final_name} running with scheduler {scheduler}")
asyncio_run(run.loop())
```
## STOP RUNNING TASKS
Functions were not meant to be scheduled!!! Do this immediately:
```python
from runevery import run, Scheduler
@run.once()
async def stop_everything(scheduler: Scheduler):
for task in scheduler:
task.discard()
asyncio_run(run.loop())
```
## Pause a task
If you need to suspend (pause) the task execution `task.pause_for` and `task.pause_until` are for your service
```python
from runevery import run, SchedulingTask
@run.every(minutes=5)
async def pausable_task(task: SchedulingTask):
print("Running pausable task")
task.pause_for(600) # Pause for 10 minutes
asyncio_run(run.loop())
```
## Run a task on a custom condition
Want to run a task based on a custom condition? Use a custom planner.
```python
from runevery import run, SchedulingPlanner, SchedulingTask
class CustomConditionPlanner(SchedulingPlanner):
def check(self, task: SchedulingTask):
return some_custom_condition()
@run.plan(planner=CustomConditionPlanner())
async def custom_task():
print("Running custom task")
asyncio_run(scheduler.loop())
```
## Handle errors
Need to handle errors in your tasks? Provide an `on_error` callback.
```python
from runevery import run, SchedulingTask
@run.every(minutes=5, on_error=lambda task: print(f"Error in {task.final_name}"))
async def error_prone_task():
raise Exception("Oops!")
asyncio_run(run.loop())
```
And that's it! You've now got a solid understanding of how to use runevery to schedule tasks in Python. Happy scheduling!
Raw data
{
"_id": null,
"home_page": "https://github.com/evtn/runevery",
"name": "runevery",
"maintainer": null,
"docs_url": null,
"requires_python": "<4.0,>=3.8",
"maintainer_email": null,
"keywords": null,
"author": "Dmitry Gritsenko",
"author_email": "runevery@evtn.me",
"download_url": "https://files.pythonhosted.org/packages/ce/7a/07482d60e31e77316df06d6de41aff04245891fda7a194a753963cf12ba1/runevery-0.5.2.tar.gz",
"platform": null,
"description": "# Runevery task when it's time\n\nRunevery (one word, as in 'bakery') is a simple async scheduling library designed to schedule tasks in _no time_. With a simple and intuitive API, you can plan and manage tasks to run at specific intervals, times, or even based on custom conditions. No more writing complex loops and time checks!\n\n> [!NOTE]\n> It is highly recommended to use type-checking with runevery. Not only did I spent some time polishing it here, but also types are cool and nice. And I won't be covering some stuff that is obvious from the names.\n\n> [!IMPORTANT]\n> This package is fresh and both the readme and the API are not very stable and finished. Feel free to create issues and stuff, but be aware that this is at its early stages.\n\n## Installation\n\nYou can install runevery from PyPI using pip. Just run:\n\n```sh\npip install runevery\n```\n\nOr, if you prefer Poetry:\n\n```sh\npoetry add runevery\n```\n\nFor more details, check out the [PyPI page](https://pypi.org/project/runevery/).\n\n## Basic usage\n\nLet's start with some basic usage examples to get you familiar with runevery.\n\n### Example: Fetching emails every 10 minutes\n\nSuppose you want to fetch your emails every 10 minutes. Here's how you can do it:\n\n```python\nfrom asyncio import run as asyncio_run\nfrom runevery import run\n\n@run.every(minutes=10)\nasync def fetch_emails():\n print(\"Very convincing code to fetch e-mails or something\")\n\nasyncio_run(run.loop())\n```\n\n### Example: Updating weather data every hour\n\nNeed to update your weather data every hour? No problem!\n\n```python\nfrom asyncio import run as asyncio_run\nfrom runevery import run\n\n@run.every(hours=1)\nasync def update_weather():\n print(\"Updating weather data...\")\n\nasyncio_run(run.loop())\n```\n\n### Example: Running a task once at startup\n\nSometimes, you might want to run a task just once at startup. Here's how:\n\n```python\nfrom asyncio import run as asyncio_run\nfrom runevery import run\n\n@run.once()\nasync def startup_task():\n print(\"Running startup task...\")\n\nasyncio_run(run.loop())\n```\n\n### Example: Running a task after a specific time\n\nWant to run a task after a specific Unix timestamp, just for fun? We have a tool for that, it's called `scheduler.after`:\n\n```python\nfrom asyncio import run as asyncio_run\nfrom runevery import run\n\n@run.after(offset=1718567362)\nasync def run_after_timestamp():\n print(\"Running after specific timestamp...\")\n\nasyncio_run(run.loop())\n```\n\n## What's a scheduler, anyway?\n\n### What is the scheduler task loop?\n\nThe scheduler and its task loop are the core of runevery. It continuously ticks through all scheduled tasks, checking if it's time to run them based on their planners.\n\nBut how does the scheduler know when to run a task? It doesn't. **CREDITS ROLL**\nOkay, but actually, it just ticks every task, and each task's planner decides if it's time to run.\n\n### What are the arguments to scheduler methods?\n\nLet's break down the arguments you can pass to `scheduler.plan`, `scheduler.every`, `scheduler.once`, `scheduler.at`, and `scheduler.after`.\n\n#### Example: Using `scheduler.every`\n\n```python\nfrom asyncio import run as asyncio_run\nfrom runevery import run\n\n@run.every(seconds=30)\nasync def every_30_seconds():\n print(\"Running every 30 seconds\")\n\nasyncio_run(run.loop())\n```\n\n#### Example: Using `scheduler.at`\n\n```python\nfrom asyncio import run as asyncio_run\nfrom runevery import run\n\n@run.at(interval=86400, hours=15)\nasync def daily_at_3pm():\n print(\"Running daily at 3 PM\")\n\nasyncio_run(run.loop())\n```\n\n## What's a planner?\n\nPlanners are the brains behind the scheduling. They decide if a task should run based on various conditions.\n\n### Built-in planners\n\n#### IntervalPlanner\n\nRuns tasks at fixed intervals.\n\n```python\nfrom runevery import IntervalPlanner, run\n\n\n@run.plan(planner=IntervalPlanner(interval=60))\nasync def every_minute():\n print(\"Running every minute\")\n\nasyncio_run(run.loop())\n```\n\n#### FixedOffsetPlanner\n\nRuns tasks at a fixed timestamp, optionally repeating at intervals.\n\n```python\nfrom runevery import FixedOffsetPlanner, Scheduler\n\nscheduler = Scheduler()\n\n@scheduler.plan(planner=FixedOffsetPlanner(offset=1718567362, interval=3600))\nasync def hourly_after_timestamp():\n print(\"Running hourly after a specific timestamp\")\n\nasyncio_run(scheduler.loop())\n```\n\n#### CooldownPlanner\n\nRuns tasks based on a cooldown period.\n\n```python\nfrom runevery import CooldownPlanner, run\n\nclass MyCooldownSource:\n def get_cooldown(self, interval: float):\n return 0 # Replace with actual cooldown logic\n\n@run.plan(planner=CooldownPlanner(MyCooldownSource(), interval=300))\nasync def cooldown_task():\n print(\"Running with cooldown\")\n\nasyncio_run(run.loop())\n```\n\n## Interval strategy\n\nThe interval strategy defines if the interval is counted from the start or the end of the run.\n\n- Use `\"start\"` (default), when you want to count interval from the start of the previous task. The duration between callback starts doesn't drift, since it doesn't depend on the task duration.\n- Use `\"end\"`, when you care about interval _between_ tasks. This interval starts counting after the previous task has ended, so the pause between tasks doesn't drift.\n\n```python\nfrom runevery import run\nfrom asyncio import sleep\n\n@run.every(seconds=10, interval_strategy=\"end\")\nasync def very_long_task():\n await sleep(3600)\n print(\"This task will run each ~3610 seconds, despite the interval beint 10 seconds\")\n\nasyncio_run(run.loop())\n```\n\n## Arguments to the task callback\n\nTask callbacks can receive optional arguments like `task` and `scheduler`.\n\n```python\nfrom runevery import run, Scheduler, SchedulingTask\n\n@run.every(minutes=5)\nasync def task_with_args(task: SchedulingTask, scheduler: Scheduler):\n print(f\"Task {task.final_name} running with scheduler {scheduler}\")\n\nasyncio_run(run.loop())\n```\n\n## STOP RUNNING TASKS\n\nFunctions were not meant to be scheduled!!! Do this immediately:\n\n```python\nfrom runevery import run, Scheduler\n\n@run.once()\nasync def stop_everything(scheduler: Scheduler):\n for task in scheduler:\n task.discard()\n\nasyncio_run(run.loop())\n```\n\n## Pause a task\n\nIf you need to suspend (pause) the task execution `task.pause_for` and `task.pause_until` are for your service\n\n```python\nfrom runevery import run, SchedulingTask\n\n@run.every(minutes=5)\nasync def pausable_task(task: SchedulingTask):\n print(\"Running pausable task\")\n task.pause_for(600) # Pause for 10 minutes\n\nasyncio_run(run.loop())\n```\n\n## Run a task on a custom condition\n\nWant to run a task based on a custom condition? Use a custom planner.\n\n```python\nfrom runevery import run, SchedulingPlanner, SchedulingTask\n\nclass CustomConditionPlanner(SchedulingPlanner):\n def check(self, task: SchedulingTask):\n return some_custom_condition()\n\n@run.plan(planner=CustomConditionPlanner())\nasync def custom_task():\n print(\"Running custom task\")\n\nasyncio_run(scheduler.loop())\n```\n\n## Handle errors\n\nNeed to handle errors in your tasks? Provide an `on_error` callback.\n\n```python\nfrom runevery import run, SchedulingTask\n\n@run.every(minutes=5, on_error=lambda task: print(f\"Error in {task.final_name}\"))\nasync def error_prone_task():\n raise Exception(\"Oops!\")\n\nasyncio_run(run.loop())\n```\n\nAnd that's it! You've now got a solid understanding of how to use runevery to schedule tasks in Python. Happy scheduling!\n",
"bugtrack_url": null,
"license": "MIT",
"summary": null,
"version": "0.5.2",
"project_urls": {
"Homepage": "https://github.com/evtn/runevery",
"Repository": "https://github.com/evtn/runevery"
},
"split_keywords": [],
"urls": [
{
"comment_text": "",
"digests": {
"blake2b_256": "14ee32b92d8d3c07ed209de065ff0e07492f5dcaf51758bba775b910b04df65c",
"md5": "10647b90fa5e90dd4bf0633bdc5e2314",
"sha256": "f7ff25386f58810617a5ed467f408b104cccc0d5ca008c0d3d5a8aa64cdebbab"
},
"downloads": -1,
"filename": "runevery-0.5.2-py3-none-any.whl",
"has_sig": false,
"md5_digest": "10647b90fa5e90dd4bf0633bdc5e2314",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": "<4.0,>=3.8",
"size": 11202,
"upload_time": "2024-06-21T22:41:45",
"upload_time_iso_8601": "2024-06-21T22:41:45.130169Z",
"url": "https://files.pythonhosted.org/packages/14/ee/32b92d8d3c07ed209de065ff0e07492f5dcaf51758bba775b910b04df65c/runevery-0.5.2-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": "",
"digests": {
"blake2b_256": "ce7a07482d60e31e77316df06d6de41aff04245891fda7a194a753963cf12ba1",
"md5": "2a643ff95bf6c5c294f9be990f5266e1",
"sha256": "0b9788dd8e6822cd7513b8bb88aa4be8b07352eb8a84d1d015b2c86bd105caef"
},
"downloads": -1,
"filename": "runevery-0.5.2.tar.gz",
"has_sig": false,
"md5_digest": "2a643ff95bf6c5c294f9be990f5266e1",
"packagetype": "sdist",
"python_version": "source",
"requires_python": "<4.0,>=3.8",
"size": 11310,
"upload_time": "2024-06-21T22:41:46",
"upload_time_iso_8601": "2024-06-21T22:41:46.200716Z",
"url": "https://files.pythonhosted.org/packages/ce/7a/07482d60e31e77316df06d6de41aff04245891fda7a194a753963cf12ba1/runevery-0.5.2.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2024-06-21 22:41:46",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "evtn",
"github_project": "runevery",
"travis_ci": false,
"coveralls": true,
"github_actions": true,
"lcname": "runevery"
}
JSON
