Added schedule_id label to the task. (#320)

Author: s3rius

docs/guide/scheduling-tasks.md

@@ -168,3 +168,12 @@ Or it can be done manually, by calling `delete_schedule` on schedule source prov
 ```python
     await redis_source.delete_schedule(schedule.schedule_id)
 ```
+
+Also, you can get schedule_id from the tasks's labels.
+
+```python
+@broker.task
+async def my_task(context: Context = TaskiqDepends()) -> None:
+    schedule_id = context.message.labels.get("schedule_id")
+    print("Schedule ID:", schedule_id)
+```

taskiq/cli/scheduler/run.py

@@ -151,9 +151,10 @@ async def run_scheduler_loop(scheduler: TaskiqScheduler) -> None:
                     task_delay = get_task_delay(task)
                 except ValueError:
                     logger.warning(
-                        "Cannot parse cron: %s for task: %s",
+                        "Cannot parse cron: %s for task: %s, schedule_id: %s",
                         task.cron,
                         task.task_name,
+                        task.schedule_id,
                     )
                     continue
                 if task_delay is not None:

taskiq/scheduler/scheduler.py

@@ -46,7 +46,9 @@ async def on_ready(self, source: "ScheduleSource", task: ScheduledTask) -> None:
         except ScheduledTaskCancelledError:
             logger.info("Scheduled task %s has been cancelled.", task.task_name)
         else:
-            await AsyncKicker(task.task_name, self.broker, task.labels).kiq(
+            await AsyncKicker(task.task_name, self.broker, task.labels).with_labels(
+                schedule_id=task.schedule_id,
+            ).kiq(
                 *task.args,
                 **task.kwargs,
             )