SyncQueue reference¶

def close(self) -> None:
    """Stop the background event loop and join its thread.

    Safe to call more than once.
    """
    if self._closed:
        return
    self._closed = True
    self._loop.call_soon_threadsafe(self._loop.stop)
    self._thread.join(timeout=5)

def add(self, task: Task[R] | list[Task[R]]) -> Task[R] | None | list[Task[R] | None]:
    """Add task into the queue. Accept list of tasks for batch add.

    Args:
        task: Instance of a Task, or list of Task instances for batch mode.

    Returns:
        The added task, or None if its ID already exists. In batch mode,
        returns a list in input order with None for tasks that already exist.
    """
    if isinstance(task, list):  # mypy needs this
        return self._run_coro(self._queue.add(task))
    return self._run_coro(self._queue.add(task))

def get_task(self, task: Task[R] | UUID | str | list[Task[R] | UUID | str]) -> Task[R] | None | dict[UUID, Task[R]]:
    """Get task by id.

    Args:
        task: Instance of a Task or its ID, or list of Task instances/IDs for batch mode.

    Returns:
        Instance of a Task or None if not found.<br>In *batch mode*, returns Dictionary of Task IDs to Task instances.
    """
    return self._run_coro(self._queue.get_task(task))

def get_all_tasks(self) -> list[Task]:
    """Get all tasks, including completed/failed ones.

    Returns:
        List of all tasks
    """
    return self._run_coro(self._queue.get_all_tasks())

def get_pending(self, count: int = 1) -> list[Task]:
    """List pending tasks.

    Args:
        count: Number of pending tasks to retrieve.

    Returns:
        List of pending tasks
    """
    return self._run_coro(self._queue.get_pending(count))

def schedule(self, task: Task[R], at: datetime | timedelta) -> Task[R] | None:
    """Schedule task to be processed after certain time.

    Args:
        task: Instance of a Task
        at: When to process the task.<br>
            If timedelta is provided, it will be added to current time.<br>
            *Note: datetime must be offset-aware (i.e. have timezone info).*

    Returns:
        The scheduled task, or None if its ID already exists.
    """
    return self._run_coro(self._queue.schedule(task, at))

def get_scheduled(self) -> list[Task]:
    """List scheduled tasks.

    Returns:
        List of scheduled tasks
    """
    return self._run_coro(self._queue.get_scheduled())

def retry(self, task: Task[R] | UUID | str, at: datetime | timedelta | None = None, force: bool = False) -> Task[R] | None:
    """Retry failed task.

    Args:
        task: Instance of a Task or its ID
        at: When to retry the task.<br>
            - If None (default), retries immediately.<br>
            - If timedelta is provided, it will be added to current time.<br>
            *Note: datetime must be offset-aware (i.e. have timezone info).*
        force: If True, allows retrying even if task has completed successfully. Defaults to False.

    Returns:
        The re-queued task, or None if no task with the given ID exists.

    Raises:
        ValueError: If task has already completed successfully and force is not set to True.
        TypeError: If provided datetime is not offset-aware.
    """
    return self._run_coro(self._queue.retry(task, at, force))

def cancel(self, task: Task | UUID | str) -> Task:
    """Cancel a pending or scheduled task.

    Args:
        task: Instance of a Task or its ID.

    Returns:
        The updated Task instance with status 'cancelled'.

    Raises:
        TaskCancellationError: If the task cannot be cancelled - either it
            was already claimed by a worker, it already finished, or it
            does not exist.
    """
    return self._run_coro(self._queue.experimental.cancel(task))

def delete(self, task: Task | UUID | str) -> bool:
    """Hard-delete a finished task's stored metadata.

    Only tasks in a terminal state (completed, failed, crashed, cancelled) can be deleted.

    Args:
        task: Instance of a Task or its ID.

    Returns:
        True if the task existed and was deleted, False if it was not found.

    Raises:
        ValueError: If the task has not finished yet.
    """
    return self._run_coro(self._queue.experimental.delete(task))

def size(self) -> int:
    """Get number of pending tasks in the queue.

    Returns:
        Number of pending tasks
    """
    return self._run_coro(self._queue.size())

def clear(self) -> int:
    """Clear all tasks, including completed ones.

    Returns:
        Number of stored task entries removed.
    """
    return self._run_coro(self._queue.clear())

def add_cron(self, task: Task, cron: str) -> TaskCron | None:
    """Add a cron job to run a task on a schedule.

    Args:
        task: Instance of a Task
        cron: Cron expression string (e.g. "*/5 * * * *" to run every 5 minutes)

    Returns:
        The added cron, or None if an identical cron already exists.
    """
    return self._run_coro(self._queue.add_cron(task, cron))

def delete_cron(self, task: Task, cron: str) -> bool:
    """Delete a cron job.

    Args:
        task: Instance of a Task
        cron: Cron expression string used when adding the cron job

    Returns:
        Success boolean
    """
    return self._run_coro(self._queue.delete_cron(task, cron))

def get_crons(self) -> list[TaskCron]:
    """List all cron jobs.

    Returns:
        List of TaskCron instances
    """
    return self._run_coro(self._queue.get_crons())

def wait_for(self, task: Task[R] | UUID | str | list[Task[R] | UUID | str], timeout: float = 0) -> Task[R] | None | dict[UUID, Task[R]]:
    """Wait for task to complete and return updated task instance.

    Blocks the calling thread until the task finishes; use `timeout`
    to bound the wait.

    Args:
        task: Instance of a Task or its ID, or list of Task instances/IDs for batch mode.
        timeout: Maximum time to wait in seconds. Default is 0 (wait indefinitely).<br>
                 If timeout is reached, returns None (or partial results in batch mode).<br>
                 In batch mode, this is the maximum time to wait for all tasks to complete.<br>
                 Note: In non-batch mode, if timeout is reached and no task is found, a TimeoutError is raised.

    Returns:
        Instance of a Task or None if not found or timeout reached.<br>In batch mode, returns dictionary of Task IDs to Task instances (partial results possible on timeout).

    Raises:
        TimeoutError: If timeout is reached and no task is found (only in non-batch mode).
    """
    return self._run_coro(self._queue.wait_for(task, timeout))

def add_workflow(self, workflow: Workflow) -> WorkflowResult:
    """Add a workflow into the queue."""
    return self._run_coro(self._queue.experimental.add_workflow(workflow))

def wait_for_workflow(self, workflow: Workflow | UUID | str | list[Workflow | UUID | str], timeout: float = 0) -> Workflow | None | dict[UUID, Workflow]:
    """Wait for workflow to finish (complete or fail) and return the updated workflow.

    Args:
        workflow: Instance of a Workflow or its ID, or list of Workflow instances/IDs for batch mode.
        timeout: Maximum time to wait in seconds. Default is 0 (wait indefinitely).

    Returns:
        The finished Workflow, or None if not found.<br>In *batch mode*, returns a dictionary of Workflow IDs to finished Workflow instances.

    Raises:
        TimeoutError: If a positive timeout expires before all given workflows have finished.
    """
    return self._run_coro(self._queue.experimental.wait_for_workflow(workflow, timeout))

def get_workflow(self, workflow: Workflow | UUID | str | list[Workflow | UUID | str]) -> Workflow | None | dict[UUID, Workflow]:
    """Get workflow by id.

    Args:
        workflow: Instance of a Workflow or its ID, or list of Workflow instances/IDs for batch mode.

    Returns:
        Instance of a Workflow or None if not found.<br>In *batch mode*, returns Dictionary of Workflow IDs to Workflow instances.
    """
    return self._run_coro(self._queue.experimental.get_workflow(workflow))

def get_all_workflows(self) -> list[Workflow]:
    """Get all stored workflows, including completed/failed ones.

    Returns:
        List of all Workflow instances
    """
    return self._run_coro(self._queue.experimental.get_all_workflows())

def get_pending_workflows(self) -> list[Workflow]:
    """Get workflows that are neither completed nor errored.

    Returns:
        List of pending Workflow instances
    """
    return self._run_coro(self._queue.experimental.get_pending_workflows())

def delete_workflow(self, workflow: Workflow | UUID | str) -> bool:
    """Delete a workflow.

    Args:
        workflow: Instance of a Workflow or its ID

    Returns:
        True if the workflow existed and was deleted.
    """
    return self._run_coro(self._queue.experimental.delete_workflow(workflow))