How to elegantly use result_backend.set_progress ？？

[KitStarLee]

“
@broker.task
async def heavy_task(key: str, val: str, context: Context = TaskiqDepends()) -> str:

for i in range(10):
    # Simulate some work
    await asyncio.sleep(1)
    # Calculate progress
    progress = (i + 1) * 10
    # Send the progress
    # await context.set_progress(progress)

await asyncio.sleep(10)
.....

”

How to obtain the current task_id in the above code ？

Then set the progress using context.broker.result_backend.set_progress().

Because get_ progress returns None by default。

[abderrahimchadli]

The current task ID is context.message.task_id. Context stores the incoming TaskiqMessage; you don't need to find the worker PID or pass the ID as another task argument.

With the current progress API, the backend expects a TaskProgress object rather than a bare percentage:

import asyncio
from taskiq import Context, TaskiqDepends
from taskiq.depends.progress_tracker import TaskProgress, TaskState

@broker.task
async def heavy_task(
    key: str, val: str, context: Context = TaskiqDepends(),
) -> str:
    for i in range(10):
        await asyncio.sleep(1)
        await context.broker.result_backend.set_progress(
            context.message.task_id,
            TaskProgress(
                state=TaskState.STARTED,
                meta={"percent": (i + 1) * 10},
            ),
        )
    return val

Alternatively, inject ProgressTracker and use its set_progress(state, meta) helper; it uses that same context/task ID internally.

The producer and worker need the same progress-capable result backend. In RedisAsyncResultBackend, get_progress() returns None when no progress key exists, including before the first update or after its TTL expires. A 100% progress value isn't a substitute for checking the final result/error.

Sources: Context, ProgressTracker and TaskProgress, Redis progress storage. If you're still using the version from the original post, check its method signature before using the current example.