taskwait

Wait for a task to complete. This is a very small package and does not really warrant this level of complexity in documentation. We're using this to help us bootstrap up some ideas around python doc generation.

Reference

Wait for a task to complete.

This module is inspired by our R "logwatch" package.

Result dataclass

Result of waiting on a task.

Attributes

status (str): The final status, returned by the status() method start (float): The task start time, in seconds since the Epoch end (float): The task end time, in seconds since the Epoch

Source code in src/taskwait/__init__.py

@dataclass
class Result:
    """
    Result of waiting on a task.

    Attributes
    ----------
      status (str):
        The final status, returned by the ``status()`` method
      start (float):
        The task start time, in seconds since the Epoch
      end (float):
        The task end time, in seconds since the Epoch

    """

    status: str
    start: float
    end: float

Task

Bases: ABC

Base class for tasks.

Inherit from this class to create something suitable to pass into taskwait.

Attributes

status_waiting (set[str]): A set of statuses that are interpreted as "waiting" status_running (set[str]): A set of statuses that are interpreted as "running"

Source code in src/taskwait/__init__.py

class Task(ABC):
    """
    Base class for tasks.

    Inherit from this class to create something suitable to pass into
    taskwait.

    Attributes
    ----------
      status_waiting (set[str]):
          A set of statuses that are interpreted as "waiting"
      status_running (set[str]):
        A set of statuses that are interpreted as "running"

    """

    status_waiting: set[str]
    status_running: set[str]

    @abstractmethod
    def status(self) -> str:
        """Query for the status of the task."""
        pass  # pragma: no cover

    @abstractmethod
    def log(self) -> list[str] | None:
        """Fetch logs for the task, if available."""
        return None  # pragma: no cover

    @abstractmethod
    def has_log(self) -> bool:
        """Indicate if this task **may** produce logs (now or in future)."""
        pass  # pragma: no cover

has_log() abstractmethod

Indicate if this task may produce logs (now or in future).

Source code in src/taskwait/__init__.py

@abstractmethod
def has_log(self) -> bool:
    """Indicate if this task **may** produce logs (now or in future)."""
    pass  # pragma: no cover

log() abstractmethod

Fetch logs for the task, if available.

Source code in src/taskwait/__init__.py

@abstractmethod
def log(self) -> list[str] | None:
    """Fetch logs for the task, if available."""
    return None  # pragma: no cover

status() abstractmethod

Query for the status of the task.

Source code in src/taskwait/__init__.py

@abstractmethod
def status(self) -> str:
    """Query for the status of the task."""
    pass  # pragma: no cover

taskwait(task, *, show_log=True, progress=True, poll=1, timeout=None)

Wait for a task to complete.

Parameters:

Name	Type	Description	Default
task	Task	The task to wait on.	required
show_log	bool	Show logs, if available, while waiting?	True
progress	bool	Show a progress bar while waiting? Only shown while running if show_log is False.	True
poll	float	Period to poll for new status/logs, in seconds.	1
timeout	float | None	(float | None): Time, in seconds, to wait before throwing a TimeoutError. If None, we wait forever.	None

Source code in src/taskwait/__init__.py

def taskwait(
    task: Task,
    *,
    show_log: bool = True,
    progress: bool = True,
    poll: float = 1,
    timeout: float | None = None,
) -> Result:
    """
    Wait for a task to complete.

    Args:
      task (Task): The task to wait on.
      show_log (bool): Show logs, if available, while waiting?
      progress (bool): Show a progress bar while waiting? Only shown
        while running if `show_log` is `False`.
      poll (float): Period to poll for new status/logs, in seconds.
      timeout: (float | None): Time, in seconds, to wait before
        throwing a `TimeoutError`.  If `None`, we wait forever.

    """
    t = _RunningTask(
        task, show_log=show_log, progress=progress, poll=poll, timeout=timeout
    )
    return t.wait()