queue_test package

Submodules

queue_test.cli module

Console script for managing the command queue.

This script provides a command-line interface (CLI) to interact with the queue_test.queue module. Users can add commands to the queue and run all queued commands in priority order.

Commands:

add: Adds a new command to the queue. run: Processes and displays all commands in the queue.

Usage:
Add a command:

python cli.py add '{"cmd": "Task description", "priority": 1}'

Run the queue:

python cli.py run

queue_test.cli.add(value: str | None = <typer.models.ArgumentInfo object>)[source]

Adds a new command to the queue.

Args:
value (Optional[str]): A JSON string containing the command data.

Example: '{"cmd": "Task description", "priority": 1}'

queue_test.cli.run()[source]

Processes and displays all commands in the queue in priority order.

This function retrieves commands from the queue, prints their details, and processes them with a small delay to simulate execution.

queue_test.queue module

queue.py

This module defines a priority-based command queue system with persistence support.

Classes:

Command: Represents a command with a string and a priority level. Queue: Implements a priority queue with persistence using pickle.

Usage:

Create a Queue object to manage commands. Use add() to add commands to the queue, and next() to retrieve and remove commands in priority order.

Example:

from queue import Queue

# Create a queue
q = Queue()

# Add commands to the queue
q.add({"cmd": "Start backup", "priority": 2})
q.add({"cmd": "Restart server", "priority": 1})

# Process commands in priority order
for command in q:
    print(command.cmd, command.priority)
Details:

The queue state is persisted to a file (default: queue.pkl in the current directory), allowing the queue to retain its state across program executions. Command priorities must be integers between 1 and 10, where 1 represents the highest priority.

class queue_test.queue.Command(cmd: str, priority: int)[source]

Bases: object

Represents a command with an associated priority.

cmd: str
priority: int
class queue_test.queue.Queue(queue: Path | None = None)[source]

Bases: object

A priority-based queue that supports persistence using pickle.

add(command: dict) None[source]

Adds a new command to the queue. Commands are sorted by priority after being added.

Args:

command (dict): A dictionary with "cmd" (str) and "priority" (int, 1-10) keys.

Raises:

RuntimeError: If the priority is not between 1-10 or if the dictionary structure is incorrect.

next() Command | None[source]

Retrieves and removes the next command from the queue.

Returns:

Optional[Command]: The next command in the queue, or None if the queue is empty.

Module contents

Top-level package for Queue Test.