Class WorkQueue

ocrflux/work_queue.py:23–109

Base class defining the interface for a work queue.

class WorkQueue(abc.ABC):
    """
    Base class defining the interface for a work queue.
    """

    @abc.abstractmethod
    async def populate_queue(self, work_paths: List[str], items_per_group: int) -> None:
        """
        Add new items to the work queue. The specifics will vary depending on
        whether this is a local or S3-backed queue.

        Args:
            work_paths: Each individual path that we will process over
            items_per_group: Number of items to group together in a single work item
        """
        pass

    @abc.abstractmethod
    async def initialize_queue(self) -> int:
        """
        Load the work queue from the relevant store (local or remote)
        and initialize it for processing.

        For example, this might remove already completed work items and randomize
        the order before adding them to an internal queue.
        """
        pass

    @abc.abstractmethod
    async def is_completed(self, work_hash: str) -> bool:
        """
        Check if a work item has been completed.

        Args:
            work_hash: Hash of the work item to check

        Returns:
            True if the work is completed, False otherwise
        """
        pass

    @abc.abstractmethod
    async def get_work(self, worker_lock_timeout_secs: int = 1800) -> Optional[WorkItem]:
        """
        Get the next available work item that isn't completed or locked.

        Args:
            worker_lock_timeout_secs: Number of seconds before considering
                                      a worker lock stale (default 30 mins)

        Returns:
            WorkItem if work is available, None if queue is empty
        """
        pass

    @abc.abstractmethod
    async def mark_done(self, work_item: WorkItem) -> None:
        """