Delayed and scheduled tasks

[vvanglro]

Hi, I would like to ask if there will be any plans to add features about delays and timed tasks?

[davorrunje]

There are no plans for that. Our reasoning is that you can use a number of existing libraries to accomplish that, there is no reason to add yet another one of doing the same thing.

[Lancetnik]

Yep, we have no plans to implement tasks scheduling right in the FastStream core, but taskiq integration via FastStream plugin is on process. For now, you can integrate your application with smth like rocketry to use tasks scheduling: https://github.com/Lancetnik/Propan/blob/main/examples/scheduling/rocketry.py

[Lancetnik]

@vvanglro the job is done! Now you can use native taskiq integration to schedule your tasks:
https://github.com/taskiq-python/taskiq-faststream/tree/master

[vvanglro]

Thank you for doing such a great job!

I was looking at the project code and the kick method is taken over by the broker in faststream, but publish in the broker doesn't seem to support delayed parameters or tasks, that means this integration project only supports timed tasks at the moment?

[Lancetnik]

Taskiq supports cron tasks and time tasks both https://taskiq-python.github.io/available-components/schedule-sources.html#labelschedulesource

But for now you have no suitable interface to make smth like publish(..., delay=30)

Taskiq dynamic jobs is WIP now, so in the nearest future you'll can create delayed tasks dynamicaly with taskiq-faststream

From the FastStream side the only one decision to make it I can image like

async def delay_publish(broker, msg, *args, delay, **kwargs):
    async def _publish():
        await asyncio.sleep(delay)
        await broker.publish(msg, *args, **kwargs)
    asyncio.create_task(_publish())

But you can write such code yourself