Add a schedule for pipeline tasks #353

Activity

1. s3rius commented on Aug 28, 2024

   

   s3rius

   on Aug 28, 2024

   Member

   More actions

   Hi! To schedule a task you need to register it to the broker. I found this solution. Which works pretty good.

   from taskiq import TaskiqScheduler
   from taskiq.schedule_sources import LabelScheduleSource
   from taskiq_redis import ListQueueBroker
   from taskiq_pipelines import PipelineMiddleware, Pipeline
   
   broker = ListQueueBroker("redis://localhost").with_middlewares(PipelineMiddleware())
   scheduler = TaskiqScheduler(broker, [LabelScheduleSource(broker)])
   
   
   @broker.task
   async def one():
       print("one")
   
   
   pipe_task = Pipeline(broker, one).call_after(one)
   
   broker.register_task(
       pipe_task.kiq,
       "mypipe",
       schedule=[{"cron": "* * * * *"}],
   )

   or if you want to schedule it dynamically, then you can save registered task and then schedule it like a regular task. Here's an example:

   from taskiq import TaskiqScheduler
   import datetime
   from taskiq_redis import ListQueueBroker, RedisScheduleSource
   from taskiq_pipelines import PipelineMiddleware, Pipeline
   import asyncio
   
   broker = ListQueueBroker("redis://localhost").with_middlewares(PipelineMiddleware())
   redis_schedule_source = RedisScheduleSource("redis://localhost/1")
   scheduler = TaskiqScheduler(broker, [redis_schedule_source])
   
   
   @broker.task
   async def one():
       print("one")
   
   
   pipeline = Pipeline(broker, one).call_after(one)
   pipe_task = broker.register_task(pipeline.kiq, task_name="mypipe")
   
   
   async def main():
       await redis_schedule_source.startup()
       await pipe_task.schedule_by_time(
           redis_schedule_source,
           time=datetime.datetime.now(datetime.UTC) + datetime.timedelta(seconds=10),
       )
       await redis_schedule_source.shutdown()
   
   
   if __name__ == "__main__":
       asyncio.run(main())
2. casper-71 commented on Aug 29, 2024

   

   casper-71

   on Aug 29, 2024

   Author

   More actions

   Thank you so much!

   It worked.

   Please add this case to the documentation https://github.com/taskiq-python/taskiq-pipelines

   ❤️React with ❤️1Reacted by Pavel Kirilin
3. casper-71

   closed this as completedon Aug 29, 2024