fix: make sure that scheduler works with interval

Author: danfimov

examples/example_with_schedule_source.py

@@ -39,6 +39,9 @@
             "kwargs": {},  # type: dict[str, Any] | None, can be omitted.
             "labels": {},  # type: dict[str, Any] | None, can be omitted.
         },
+        {
+            "interval": 10, # type: int | timedelta | None
+        },
     ],
 )
 async def best_task_ever() -> None:

src/taskiq_pg/_internal/schedule_source.py

@@ -84,6 +84,7 @@ def extract_scheduled_tasks_from_broker(self) -> list[ScheduledTask]:
                             "schedule_id": str(uuid.uuid4()),
                             "cron": schedule.get("cron", None),
                             "cron_offset": schedule.get("cron_offset", None),
+                            "interval": schedule.get("interval", None),
                             "time": schedule.get("time", None),
                         },
                     )

src/taskiq_pg/aiopg/schedule_source.py

@@ -81,6 +81,7 @@ async def get_schedules(self) -> list["ScheduledTask"]:
                         "cron": schedule["cron"],
                         "cron_offset": schedule["cron_offset"],
                         "time": schedule["time"],
+                        "interval": schedule["interval"],
                     },
                 ),
             )

src/taskiq_pg/asyncpg/schedule_source.py

@@ -80,6 +80,7 @@ async def get_schedules(self) -> list["ScheduledTask"]:
                         "cron": schedule["cron"],
                         "cron_offset": schedule["cron_offset"],
                         "time": schedule["time"],
+                        "interval": schedule["interval"],
                     },
                 ),
             )

src/taskiq_pg/psqlpy/schedule_source.py

@@ -91,6 +91,7 @@ async def get_schedules(self) -> list["ScheduledTask"]:
                         "cron": schedule["cron"],
                         "cron_offset": schedule["cron_offset"],
                         "time": schedule["time"],
+                        "interval": schedule["interval"],
                     },
                 ),
             )

src/taskiq_pg/psycopg/schedule_source.py

@@ -88,6 +88,7 @@ async def get_schedules(self) -> list["ScheduledTask"]:
                             "cron": schedule["cron"],
                             "cron_offset": schedule["cron_offset"],
                             "time": schedule["time"],
+                            "interval": schedule["interval"],
                         },
                     ),
                 )

tests/integration/test_schedule_source.py

@@ -159,7 +159,7 @@ async def test_when_call_add_schedule__then_schedule_creates(
     schedule_source_class: type[PSQLPyScheduleSource | AiopgScheduleSource | AsyncpgScheduleSource],
 ) -> None:
     # Given
-    new_schedule = ScheduledTaskFactory.build(task_name="tests:added_schedule", cron="*/5 * * * *")
+    new_schedule = ScheduledTaskFactory.build(task_name="tests:added_schedule", cron="*/5 * * * *", interval=None)
     async with _get_schedule_source(schedule_source_class, broker_with_scheduled_tasks, pg_dsn) as schedule_source:
         await schedule_source.startup()