Skip to content

AsyncResult

chicory.AsyncResult 

AsyncResult(task_id: str, backend: Backend | None = None)

Bases: Generic[T]

Asynchronous result handler for task results.

Source code in src/chicory/result.py 
16
17
18
def __init__(self, task_id: str, backend: Backend | None = None) -> None:
    self.task_id = task_id
    self._backend = backend

get async 

get(
    timeout: float | None = None, poll_interval: float = 0.1
) -> T | None

Wait for and return the task result.

Polls the backend at regular intervals until the task reaches a terminal state (SUCCESS or FAILURE).

Parameters:

Name Type Description Default
timeout float | None

Maximum seconds to wait. None means wait forever.

 None
poll_interval float

Seconds between backend polls (default 0.1).

 0.1

Returns:

Type Description
T | None

The task's return value, or None if the task returned nothing.

Raises:

Type Description
TimeoutError

If timeout is reached before the task completes.
BackendNotConfiguredError

If no backend is attached.
Exception

Re-raised from the task if it ended in FAILURE.
Source code in src/chicory/result.py 
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
async def get(
    self, timeout: float | None = None, poll_interval: float = 0.1
) -> T | None:
    """Wait for and return the task result.

    Polls the backend at regular intervals until the task reaches a
    terminal state (``SUCCESS`` or ``FAILURE``).

    Args:
        timeout: Maximum seconds to wait. ``None`` means wait forever.
        poll_interval: Seconds between backend polls (default ``0.1``).

    Returns:
        The task's return value, or ``None`` if the task returned nothing.

    Raises:
        TimeoutError: If *timeout* is reached before the task completes.
        BackendNotConfiguredError: If no backend is attached.
        Exception: Re-raised from the task if it ended in ``FAILURE``.
    """
    backend = self._ensure_backend()

    elapsed = 0.0
    while True:
        result = await backend.get_result(self.task_id)

        if result:
            match result.state:
                case TaskState.SUCCESS:
                    return cast("T", result.result)
                case TaskState.FAILURE:
                    raise Exception(result.error or "Task failed")
                case _:
                    pass

        if timeout is not None and elapsed >= timeout:
            raise TimeoutError(
                f"Task {self.task_id} did not complete in {timeout}s"
            )

        await asyncio.sleep(poll_interval)
        elapsed += poll_interval

state async 

state() -> TaskState

Get the current task state.

Returns:

Type Description
TaskState

The current TaskState (e.g. PENDING, STARTED,
TaskState

SUCCESS, FAILURE, RETRY).

Raises:

Type Description
BackendNotConfiguredError

If no backend is attached.
Source code in src/chicory/result.py 
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
async def state(self) -> TaskState:
    """Get the current task state.

    Returns:
        The current ``TaskState`` (e.g. ``PENDING``, ``STARTED``,
        ``SUCCESS``, ``FAILURE``, ``RETRY``).

    Raises:
        BackendNotConfiguredError: If no backend is attached.
    """
    backend = self._ensure_backend()
    result = await backend.get_result(self.task_id)
    if result:
        return result.state
    return TaskState.PENDING

ready async 

ready() -> bool

Check if the task is complete (success or failure).

Returns:

Type Description
bool

True if the task reached SUCCESS or FAILURE.
Source code in src/chicory/result.py 
87
88
89
90
91
92
93
94
async def ready(self) -> bool:
    """Check if the task is complete (success or failure).

    Returns:
        ``True`` if the task reached ``SUCCESS`` or ``FAILURE``.
    """
    state = await self.state()
    return state in {TaskState.SUCCESS, TaskState.FAILURE}

failed async 

failed() -> bool

Check if the task has failed.

Returns:

Type Description
bool

True if the task state is FAILURE.
Source code in src/chicory/result.py 
 96
 97
 98
 99
100
101
102
103
async def failed(self) -> bool:
    """Check if the task has failed.

    Returns:
        ``True`` if the task state is ``FAILURE``.
    """
    state = await self.state()
    return state == TaskState.FAILURE