Added skip-first-run option for scheduler. (#253)

docs/guide/scheduling-tasks.md

@@ -80,3 +80,15 @@ For the `time` field of `ScheduledTask` we use timezone information from datetim
 
 For `cron` tasks, we have an additional field called `cron_offset` that can be used to specify
 an offset of the cron task. An offset can be a string like `Europe/Berlin` or an instance of the `timedelta` class.
+
+## Skipping first run
+
+By default, when you start the scheduler it will get all tasks from the schedule source and check whether they should have been executed in this minute. If tasks should have been executed, they will be executed.
+
+This behaviour might be not convinient for some developers. For example, if you have a task that should be executed on every minute, it will be executed once you start the scheduler, even if it was executed a few seconds ago.
+
+To avoid this behaviour, you can pass the `--skip-first-run` flag to the `taskiq scheduler` command. In this case, the scheduler will wait until the start of the next minute and then start executing tasks.
+
+```bash:no-line-numbers
+taskiq scheduler module:scheduler --skip-first-run
+```

taskiq/cli/scheduler/args.py

@@ -16,6 +16,7 @@ class SchedulerArgs:
     configure_logging: bool = True
     fs_discover: bool = False
     tasks_pattern: str = "tasks.py"
+    skip_first_run: bool = False
 
     @classmethod
     def from_cli(cls, args: Optional[Sequence[str]] = None) -> "SchedulerArgs":
@@ -66,4 +67,13 @@ def from_cli(cls, args: Optional[Sequence[str]] = None) -> "SchedulerArgs":
             dest="configure_logging",
             help="Use this parameter if your application configures custom logging.",
         )
+        parser.add_argument(
+            "--skip-first-run",
+            action="store_true",
+            dest="skip_first_run",
+            help=(
+                "Skip first run of scheduler. "
+                "This option skips running tasks immediately after scheduler start."
+            ),
+        )
         return cls(**parser.parse_args(args).__dict__)

taskiq/cli/scheduler/run.py

@@ -206,7 +206,15 @@ async def run_scheduler(args: SchedulerArgs) -> None:
     logger.info("Starting scheduler.")
     await scheduler.startup()
     logger.info("Startup completed.")
-
+    if args.skip_first_run:
+        next_minute = datetime.utcnow().replace(second=0, microsecond=0) + timedelta(
+            minutes=1,
+        )
+        delay = next_minute - datetime.utcnow()
+        delay_secs = int(delay.total_seconds())
+        logger.info(f"Skipping first run. Waiting {delay_secs} seconds.")
+        await asyncio.sleep(delay.total_seconds())
+        logger.info("First run skipped. The scheduler is now running.")
     try:
         await run_scheduler_loop(scheduler)
     except asyncio.CancelledError: