How to block a task until it's finished ?

[sorasful]

Hi there!

I was trying to wait for a task to finish before going further, and I thought that wait_result() was for this purpose. But it seems that it does not block. Here's my setup .

# import related stuff ...

settings = get_settings()

broker = ListQueueBroker(
    url=settings.redis.URL,
)

scheduler = TaskiqScheduler(
    broker=broker,
    sources=[LabelScheduleSource(broker)],
)


@broker.task(task_name="process_video")
async def process_video(video_id: uuid.UUID):
    """Main task to process a YouTube video."""
    print('WAITING')
    await asyncio.sleep(10)
    print("DONE WAITING")
    return



if __name__ == "__main__":
    async def main() -> None:
        process_task = await process_video.kiq(video_id="hdd")
        await process_task.wait_result()
    
        print("DONE")

The thing is that my task is indeed triggered and takes 10 seconds. But I see the "DONE" print immediately, I would know if it's possible to block it until it's done, I tried to setup timeout params but it's the same.

Thanks for your time !

[ssilaev]

First you need to start the loop and the broker.

Try this:

@broker.task(task_name="process_video")
async def process_video(video_id: uuid.UUID):
    """Main task to process a YouTube video."""
    print('WAITING')
    await asyncio.sleep(10)
    print("DONE WAITING")
    return

async def async_main() -> None:
    await broker.startup()

    process_task = await process_video.kiq(video_id=uuid.uuid4())
    await process_task.wait_result()
    print("DONE")

    await broker.shutdown()

if __name__ == "__main__":
    asyncio.run(async_main())

[sorasful]

@enodllew Yes, it's been done, but for the sake of simplicity I did not put it there. The task is working nicely, I just want a wait to wait before going further in my code.

[ssilaev]

@enodllew Yes, it's been done, but for the sake of simplicity I did not put it there. The task is working nicely, I just want a wait to wait before going further in my code.

Ah, I see, sry. Then I don't quite understand what kind of result you're expecting. I hope someone else will be able to help you.

[sorasful]

No worries.

My problem is that I was thinking that await process_task.wait_result() would be blocking until we have the result of the task. But it seems it is not the way to do this.

Edit: I've added this :

r = await process_task.wait_result(with_logs=True)
print(f"Task execution took: {r.execution_time} seconds.")

And I got

Task execution took: 0.0 seconds.

But actually the task is running in background. So I wonder what can be the problem.

[s3rius]

@sorasful, process_task.wait_result works as you would expect and blocks until task is ready. BUT it depends on result backend. And unless you have it, it won't wait at all. So, in order to resolve the issue, connect a result backend.

[sorasful]

@s3rius Thank you for the quick answer, and it was exactly this. Maybe we should emphasize this in the documentation or put a warning when we use a .wait_for with not Result Backend setup ?

I'm closing this anyway, thanks again!

[s3rius]

I guess warning would be more appropriate. Let me make an issue for that.