Task

Bases: Generic[Param, RValue]

Represent a task that is not yet queued to be executed. It is not constructed by the user, but it is returned when calling a task function.

Source code in src/alchemical_queues/tasks/main.py

class Task(Generic[Param, RValue]):
    """Represent a task that is not yet queued to be executed. It is not
    constructed by the user, but it is returned when calling a task function."""

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

    def schedule(
        self,
        on_queue: AlchemicalTaskQueue,
        *,
        schedule_at: Union[datetime, None] = None,
        priority: int = 0,
        max_retries: int = 0,
        retry_in: Union[timedelta, None] = None,
    ) -> QueuedTask[RValue]:
        """Schedule a task on a queue to be executed.

        Args:
            on_queue (AlchemicalTaskQueue): the queue used as task queue.
                                        You are expected to run a worker 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 = 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 QueuedTask(queue=on_queue, entry_id=entry.entry_id, name=name)

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

Schedule a task on a queue to be executed.

Parameters:

Name	Type	Description	Default
on_queue	AlchemicalTaskQueue	the queue used as task queue. You are expected to run a worker 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/main.py

def schedule(
    self,
    on_queue: AlchemicalTaskQueue,
    *,
    schedule_at: Union[datetime, None] = None,
    priority: int = 0,
    max_retries: int = 0,
    retry_in: Union[timedelta, None] = None,
) -> QueuedTask[RValue]:
    """Schedule a task on a queue to be executed.

    Args:
        on_queue (AlchemicalTaskQueue): the queue used as task queue.
                                    You are expected to run a worker 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 = 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 QueuedTask(queue=on_queue, entry_id=entry.entry_id, name=name)