Skip to content

Task Results & Waiting

Every call to await queue_service.enqueue(...) returns a TaskResult[T] handle typed to the task function's return annotation T.

Runnable Example

docs/examples/learn/results_and_wait.py
"""Waiting for task completion and refreshing TaskResult handles."""

from litestar_queues import QueueConfig, QueueService, TaskExitCode, WorkerConfig, consume_one, task


@task("analytics.compute_summary", queue="analytics")
async def compute_summary(values: list[int]) -> dict[str, int]:
    """Compute summary statistics for a batch of integers."""
    return {"count": len(values), "total": sum(values)}


async def main() -> None:
    """Enqueue a task, claim it with consume_one, and await the TaskResult."""
    config = QueueConfig(
        queue_backend="memory",
        execution_backend="local",
        worker=WorkerConfig(placement="external", queues=("analytics",)),
    )
    async with QueueService(config=config) as service:
        handle = await service.enqueue(compute_summary, [10, 20, 30])
        assert handle.status == "pending"

        exit_code = await consume_one(service, handle.id)
        assert exit_code == TaskExitCode.SUCCESS

        completed = await handle.wait(timeout=2.0)
        assert completed.status == "completed"
        assert completed.result == {"count": 3, "total": 60}
        assert completed.error is None

Inspecting TaskResult Handles

A TaskResult wraps the most recently fetched QueuedTaskRecord and exposes convenience properties. Because TaskResult caches the snapshot from enqueue(), it does not mutate automatically in the background when a remote worker finishes the task. Refresh it after receiving a terminal event (via await handle.refresh()) or call await handle.wait(...) to fetch the terminal record:

Property / Method Return Type Description
handle.id UUID Unique task identifier (QueuedTaskRecord.id)
handle.status TaskStatus Cached status (pending, scheduled, running, completed, failed, cancelled, expired)
handle.result T \| None Cached return value once status == "completed"
handle.error str \| None Formatted error string from the latest failed attempt
handle.record QueuedTaskRecord \| None Full underlying queue record
await handle.refresh() Self Reloads QueuedTaskRecord from the storage backend in place
await handle.wait(...) Self Waits for a terminal status (completed, failed, cancelled, expired), refreshes handle.record in place, and returns self (or raises TimeoutError if timeout elapses)

Waiting for Completion

await handle.wait(timeout=10.0, poll_interval=0.1) blocks until the task reaches a terminal state (completed, failed, cancelled, or expired) and returns the refreshed TaskResult handle:

  1. Push-Accelerated Waiting: When the storage backend advertises capabilities.supports_completion_events == True (such as InMemoryQueueBackend, RedisQueueBackend, ValkeyQueueBackend, or SQLSpecQueueBackend on PostgreSQL with notify / notify_queue), wait() subscribes to completion notifications and wakes immediately when the task finishes.
  2. Polling Fallback: On backends without completion push notifications, wait() polls backend.get_task(task_id) every poll_interval seconds until the record becomes terminal or timeout elapses.

Outcomes & Exceptions from Wait

Condition Behavior of await handle.wait()
status == "completed" Returns refreshed handle (handle.result holds deserialized T)
status in {"failed", "cancelled", "expired"} Returns refreshed handle (handle.status and handle.error reflect the terminal state)
handle.record is None before waiting Raises RuntimeError("Task record is not available")
timeout elapsed while non-terminal Raises TimeoutError

Fetching Records by Task ID

When an HTTP endpoint or CLI command only has a UUID (for example from a URL path parameter), look up the record directly through QueueService.get_task(task_id):

record = await queue_service.get_task(task_id)
if record is not None and record.is_terminal:
    print(record.status, record.result)