Skip to content

AsyncTask

Bases: Generic[Param, RValue]

Async equivalent of tasks.Task. Not constructed by the user, but returned when calling an async_task function.

Source code in src/alchemical_queues/tasks/aio.py 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
class AsyncTask(Generic[Param, RValue]):
    """Async equivalent of [tasks.Task][alchemical_queues.tasks.Task]. Not
    constructed by the user, but returned when calling an `async_task`
    function."""

    def __init__(
        self,
        handler: Callable[Concatenate[TaskInfo, Param], Awaitable[RValue]],
        *args: Param.args,
        **kwargs: Param.kwargs,
    ):
        self._handler = handler
        self._args = args
        self._kwargs = kwargs

    async def schedule(
        self,
        on_queue: AsyncAlchemicalTaskQueue,
        *,
        schedule_at: Union[datetime, None] = None,
        priority: int = 0,
        max_retries: int = 0,
        retry_in: Union[timedelta, None] = None,
    ) -> AsyncQueuedTask[RValue]:
        """Schedule a task on a queue to be executed. See
        [Task.schedule][alchemical_queues.tasks.Task.schedule] for details.

        Args:
            on_queue (AsyncAlchemicalTaskQueue): the queue used as task queue.
                You are expected to run an `AsyncWorker` connected to this queue.
            schedule_at (datetime, optional): do not run the task before this time.
            priority (int, optional): the task priority, using normal priority queue semantics.
            max_retries (int, optional): how many times the task should be retried before reporting failure.
            retry_in (timedelta, optional): the minimal timespan between two tries.
        """
        name = f"{self._handler.__module__}.{self._handler.__qualname__}"
        entry = await on_queue.put(
            {
                "function": name,
                "args": self._args,
                "kwargs": self._kwargs,
                "retries": 0,
                "retry_in": retry_in,
                "max_retries": max_retries,
            },
            schedule_at=schedule_at,
            priority=priority,
        )
        return AsyncQueuedTask(queue=on_queue, entry_id=entry.entry_id, name=name)

schedule(on_queue, *, schedule_at=None, priority=0, max_retries=0, retry_in=None) async

Schedule a task on a queue to be executed. See Task.schedule for details.

Parameters:

Name Type Description Default
on_queue AsyncAlchemicalTaskQueue

the queue used as task queue. You are expected to run an AsyncWorker connected to this queue.

 required
schedule_at datetime

do not run the task before this time.

 None
priority int

the task priority, using normal priority queue semantics.

 0
max_retries int

how many times the task should be retried before reporting failure.

 0
retry_in timedelta

the minimal timespan between two tries.

 None
Source code in src/alchemical_queues/tasks/aio.py 
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
async def schedule(
    self,
    on_queue: AsyncAlchemicalTaskQueue,
    *,
    schedule_at: Union[datetime, None] = None,
    priority: int = 0,
    max_retries: int = 0,
    retry_in: Union[timedelta, None] = None,
) -> AsyncQueuedTask[RValue]:
    """Schedule a task on a queue to be executed. See
    [Task.schedule][alchemical_queues.tasks.Task.schedule] for details.

    Args:
        on_queue (AsyncAlchemicalTaskQueue): the queue used as task queue.
            You are expected to run an `AsyncWorker` connected to this queue.
        schedule_at (datetime, optional): do not run the task before this time.
        priority (int, optional): the task priority, using normal priority queue semantics.
        max_retries (int, optional): how many times the task should be retried before reporting failure.
        retry_in (timedelta, optional): the minimal timespan between two tries.
    """
    name = f"{self._handler.__module__}.{self._handler.__qualname__}"
    entry = await on_queue.put(
        {
            "function": name,
            "args": self._args,
            "kwargs": self._kwargs,
            "retries": 0,
            "retry_in": retry_in,
            "max_retries": max_retries,
        },
        schedule_at=schedule_at,
        priority=priority,
    )
    return AsyncQueuedTask(queue=on_queue, entry_id=entry.entry_id, name=name)