Skip to content

AsyncQueuedTask

Bases: Generic[RValue]

Async equivalent of tasks.QueuedTask.

Attributes:

Name Type Description
entry_id int

The id of the entry into the queue that contains the task description.
Source code in src/alchemical_queues/tasks/aio.py 
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
70
71
72
73
74
75
76
77
78
79
80
81
82
class AsyncQueuedTask(Generic[RValue]):
    """Async equivalent of [tasks.QueuedTask][alchemical_queues.tasks.QueuedTask].

    Attributes:
        entry_id (int): The id of the entry into the queue that contains the task description.
    """

    def __init__(self, queue: AsyncAlchemicalTaskQueue, entry_id: int, name: str):
        self._queue = queue
        self.entry_id = entry_id
        self._name = name

    async def done(self) -> bool:
        """Whether this task has a recorded outcome yet. See
        [QueuedTask.done][alchemical_queues.tasks.QueuedTask.done] for
        details.
        """
        return bool(await self._queue.responses(self.entry_id))

    async def result(self) -> Union[RValue, TaskException, None]:
        """Obtain the result of a queued task if it is finished, an
        exception if the task failed to run, or None if the task has not
        completed. See
        [QueuedTask.result][alchemical_queues.tasks.QueuedTask.result] for
        details.
        """
        responses = await self._queue.responses(self.entry_id)

        if not responses:
            return None

        data = responses[0].data

        if not isinstance(data, dict):
            return cast(RValue, data)

        if "error" in data:
            return TaskException(data["error"], data.get("error_type"))

        return cast(RValue, data.get("result"))

done() async

Whether this task has a recorded outcome yet. See QueuedTask.done for details.

Source code in src/alchemical_queues/tasks/aio.py 
55
56
57
58
59
60
async def done(self) -> bool:
    """Whether this task has a recorded outcome yet. See
    [QueuedTask.done][alchemical_queues.tasks.QueuedTask.done] for
    details.
    """
    return bool(await self._queue.responses(self.entry_id))

result() async

Obtain the result of a queued task if it is finished, an exception if the task failed to run, or None if the task has not completed. See QueuedTask.result for details.

Source code in src/alchemical_queues/tasks/aio.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
async def result(self) -> Union[RValue, TaskException, None]:
    """Obtain the result of a queued task if it is finished, an
    exception if the task failed to run, or None if the task has not
    completed. See
    [QueuedTask.result][alchemical_queues.tasks.QueuedTask.result] for
    details.
    """
    responses = await self._queue.responses(self.entry_id)

    if not responses:
        return None

    data = responses[0].data

    if not isinstance(data, dict):
        return cast(RValue, data)

    if "error" in data:
        return TaskException(data["error"], data.get("error_type"))

    return cast(RValue, data.get("result"))