fix: recursive dict in labels during schedule discovery in LabelScheduleSource

Author: danfimov

taskiq/schedule_sources/label_based.py

@@ -1,4 +1,5 @@
 import uuid
+from copy import deepcopy
 from logging import getLogger
 from typing import Dict, List
 
@@ -46,8 +47,12 @@ async def startup(self) -> None:
                 if "cron" not in schedule and "time" not in schedule:
                     continue
                 labels = schedule.get("labels", {})
-                labels.update(task.labels)
-                schedule_id = uuid.uuid4().hex
+
+                task_labels = deepcopy(task.labels)
+                task_labels.pop("schedule", None)
+
+                labels.update(task_labels)
+                schedule_id = schedule.get("schedule_id", uuid.uuid4().hex)
 
                 self.schedules[schedule_id] = ScheduledTask(
                     task_name=task_name,

tests/schedule_sources/test_label_based.py

@@ -20,6 +20,14 @@
     [
         pytest.param([{"cron": "* * * * *"}], id="cron"),
         pytest.param([{"time": datetime.now(pytz.UTC)}], id="time"),
+        pytest.param(
+            [{"time": datetime.now(pytz.UTC), "labels": {"foo": "bar"}}],
+            id="labels_inside_schedule",
+        ),
+        pytest.param(
+            [{"cron": "*/1 * * * *", "schedule_id": "every_minute"}],
+            id="schedule_with_id",
+        ),
     ],
 )
 async def test_label_discovery(schedule_label: List[Dict[str, Any]]) -> None:
@@ -37,16 +45,20 @@ def task() -> None:
     schedules = await source.get_schedules()
     assert schedules == [
         ScheduledTask(
-            schedule_id=schedules[0].schedule_id,
+            schedule_id=schedule_label[0].get("schedule_id", schedules[0].schedule_id),
             cron=schedule_label[0].get("cron"),
             time=schedule_label[0].get("time"),
             task_name="test_task",
-            labels={"schedule": schedule_label},
+            labels=schedule_label[0].get("labels", {}),
             args=[],
             kwargs={},
         ),
     ]
 
+    # check that labels of tasks are not changed after startup and discovery process
+    task_from_broker = next(iter(broker.get_all_tasks().values()))
+    assert task_from_broker.labels == {"schedule": schedule_label}
+
 
 @pytest.mark.anyio
 async def test_label_discovery_no_cron() -> None: