hyperion.adapters.queue.filesystem

hyperion.adapters.queue.filesystem

File-backed :class:Queue adapter (lite -- stdlib only).

FileQueue

FileQueue(queue_path, *, overwrite=False, exclusive=False)

Bases: Queue

File-based message queue.

If the queue is exclusive and it is entered multiple times, a RuntimeError will be raised.

Parameters:

Name	Type	Description	Default
queue_path	Path	The path to the queue file.	required
overwrite	bool	Whether to overwrite the queue file if it exists. Defaults to False.	False
exclusive	bool	Whether to use exclusive access to the queue file. Defaults to False.	False

Source code in hyperion/adapters/queue/filesystem.py

def __init__(self, queue_path: Path, *, overwrite: bool = False, exclusive: bool = False) -> None:
    """File-based message queue.

    If the queue is exclusive and it is entered multiple times, a RuntimeError will be raised.

    Args:
        queue_path (Path): The path to the queue file.
        overwrite (bool, optional): Whether to overwrite the queue file if it exists. Defaults to False.
        exclusive (bool, optional): Whether to use exclusive access to the queue file. Defaults to False.
    """
    self._messages: list[Message] = []
    self._context_stacklevel = 0
    self._exclusive = exclusive
    self.queue_path = queue_path.resolve()
    super().__init__()
    if queue_path.exists():
        if not queue_path.resolve().is_file():
            raise FileExistsError(f"Queue path {queue_path} already exists and is not a file.")
        if not overwrite:
            raise FileExistsError(f"Queue path {queue_path} already exists, set overwrite=True to ignore this.")
        logger.warning("Queue path already exists, it will be overwritten.", queue_path=queue_path, queue=self)