Add intervals for scheduled task (#498)

* add scheduler intervals

* add some tests, interval value validation

* tests fixes

* resolve conflicts

* tests fixes

* extract interval validation to separate module

* change schedules storage to list of tuples

---------

Co-authored-by: Pavel Kirilin <[email protected]>

Author: so-saf

docs/available-components/schedule-sources.md

@@ -37,9 +37,10 @@ The format of the schedule label is the following:
 @broker.task(
     schedule=[
         {
-            "cron": "*/1 * * * *", # type: str, either cron or time should be specified.
-            "cron_offset": None # type: str | timedelta | None, can be omitted.
-            "time": None  # type: datetime | None, either cron or time should be specified.
+            "cron": "* * * * *", # type: str, either cron, interval or time should be specified.
+            "cron_offset": None, # type: str | timedelta | None, can be omitted.
+            "interval": None,  # type: int | timedelta, either cron, interval or time should be specified.
+            "time": None,  # type: datetime | None, either cron, interval or time should be specified.
             "args": [], # type List[Any] | None, can be omitted.
             "kwargs": {}, # type: Dict[str, Any] | None, can be omitted.
             "labels": {}, # type: Dict[str, Any] | None, can be omitted.
@@ -55,6 +56,7 @@ Parameters:
 
 - `cron` - crontab string when to run the task.
 - `cron_offset` - timezone offset for cron values. Explained [here](../guide/scheduling-tasks.md#working-with-timezones)
+- `interval` - Interval to run periodic tasks. Must be at least 1 second and cannot have fractional seconds.
 - `time` - specific time when send the task.
 - `args` - args to use, when invoking the task.
 - `kwargs` - key-word arguments to use when invoking the task.

docs/guide/scheduling-tasks.md

@@ -124,7 +124,7 @@ Now we can use this source to add new schedules in runtime. Here's an example:
     )
 ```
 
-Or if you want to use cron schedules instead, just use `schedule_by_cron` method.
+You can also use cron or interval scheduling, just use the `schedule_by_cron` or `schedule_by_interval` methods
 
 ```python
     await my_task.schedule_by_cron(
@@ -135,6 +135,15 @@ Or if you want to use cron schedules instead, just use `schedule_by_cron` method
     )
 ```
 
+```python
+    await my_task.schedule_by_interval(
+        redis_source,
+        datetime.timedelta(seconds=5),
+        11,
+        arg2="arg2",
+    )
+```
+
 If you want to pass additional labels, you can call these methods on the `Kicker` instance.
 
 ```python

taskiq/api/scheduler.py

@@ -1,14 +1,15 @@
 from datetime import timedelta
 from typing import Optional
 
-from taskiq.cli.scheduler.run import run_scheduler_loop
+from taskiq.cli.scheduler.run import SchedulerLoop
 from taskiq.scheduler.scheduler import TaskiqScheduler
 
 
 async def run_scheduler_task(
     scheduler: TaskiqScheduler,
     run_startup: bool = False,
     interval: Optional[timedelta] = None,
+    loop_interval: Optional[timedelta] = None,
 ) -> None:
     """
     Run scheduler task.
@@ -18,10 +19,16 @@ async def run_scheduler_task(
 
     :param scheduler: scheduler instance.
     :param run_startup: whether to run startup function or not.
+    :param interval: interval to check for schedule updates.
+    :param loop_interval: interval to check tasks to send.
     """
     for source in scheduler.sources:
         await source.startup()
     if run_startup:
         await scheduler.startup()
     while True:
-        await run_scheduler_loop(scheduler, interval)
+        scheduler_loop = SchedulerLoop(scheduler)
+        await scheduler_loop.run(
+            update_interval=interval,
+            loop_interval=loop_interval,
+        )

taskiq/cli/scheduler/args.py

@@ -19,6 +19,7 @@ class SchedulerArgs:
     tasks_pattern: Sequence[str] = ("**/tasks.py",)
     skip_first_run: bool = False
     update_interval: Optional[int] = None
+    loop_interval: Optional[int] = None
 
     @classmethod
     def from_cli(cls, args: Optional[Sequence[str]] = None) -> "SchedulerArgs":
@@ -101,6 +102,15 @@ def from_cli(cls, args: Optional[Sequence[str]] = None) -> "SchedulerArgs":
                 "If not specified, scheduler will run once a minute."
             ),
         )
+        parser.add_argument(
+            "--loop-interval",
+            type=int,
+            default=None,
+            help=(
+                "Interval in seconds to check tasks to send. "
+                "If not specified, scheduler will run once a second."
+            ),
+        )
 
         namespace = parser.parse_args(args)
         # If there are any patterns specified, remove default.