Rethink the CLI commands

There are two main sections. One for doing the main operations, monitoring and running invocations, and the other for describing command related to entities.

Create a new sync command that ensures the queues are on the broker, and don't auto-create queues. A future iteration will allow for creating a lockfile and also be able to drop unneeded queues from the broker.

Author: ryanhiebert

README.md

@@ -58,6 +58,12 @@ to allow a project to be deployed in multiple environments.
 QUEUEIO_PIKA='amqp://guest:guest@localhost:5672/'
 ```
 
+Sync the queues to the broker:
+
+```python
+queueio sync
+```
+
 Submit the routine to run on a worker:
 
 ```python
@@ -71,7 +77,7 @@ with activate():
 Then run the worker to process submitted routines:
 
 ```sh
-queueio worker queueio=3
+queueio run queueio=3
 ```
 
 Monitor the status of active routine invocations:

queueio/__main__.py

@@ -10,9 +10,25 @@
 
 app = Typer()
 
-
-@app.command()
-def routines():
+routine_app = Typer()
+queue_app = Typer()
+
+app.add_typer(
+    routine_app,
+    name="routine",
+    help="A function to coordinate background execution.",
+    rich_help_panel="Entities",
+)
+app.add_typer(
+    queue_app,
+    name="queue",
+    help="An ordered collection of work items to process.",
+    rich_help_panel="Entities",
+)
+
+
+@routine_app.command("list")
+def routine_list():
     """Show all registered routines."""
     queueio = QueueIO()
     try:
@@ -39,11 +55,11 @@ def routines():
         queueio.shutdown()
 
 
-@app.command()
+@app.command(rich_help_panel="Commands")
 def monitor(raw: bool = False):
     """Monitor queueio events.
 
-    Shows a live view of queueio activity. Use --raw for detailed event output.
+    Show a live view of queueio activity. Use --raw for detailed event output.
     """
     if raw:
         queueio = QueueIO()
@@ -59,8 +75,8 @@ def monitor(raw: bool = False):
         Monitor().run()
 
 
-@app.command()
-def worker(
+@app.command(rich_help_panel="Commands")
+def run(
     queuespec: Annotated[
         QueueSpec,
         Argument(
@@ -71,7 +87,7 @@ def worker(
         ),
     ],
 ):
-    """Start a worker to process from a queue.
+    """Run a worker to process from a queue.
 
     The worker will process invocations from the specified queue,
     as many at a time as specified by the concurrency.
@@ -80,8 +96,31 @@ def worker(
     Worker(queueio, queuespec)()
 
 
-@app.command()
-def purge(
+@app.command(rich_help_panel="Commands")
+def sync():
+    """Sync known queues to the broker."""
+    queueio = QueueIO()
+    try:
+        routines = queueio.routines()
+
+        if not routines:
+            print("No routines registered.")
+            return
+
+        queues = sorted({routine.queue for routine in routines})
+
+        print(f"Syncing queues for {len(routines)} routine(s):")
+        for queue in queues:
+            print(f"  Ensuring queue exists: {queue}")
+            queueio.create(queue=queue)
+
+        print(f"Successfully synced {len(queues)} queue(s)")
+    finally:
+        queueio.shutdown()
+
+
+@queue_app.command("purge")
+def queue_purge(
     queues: Annotated[
         str,
         Argument(

queueio/broker.py

@@ -24,6 +24,11 @@ def enqueue(self, body: bytes, /, *, queue: str):
         """Enqueue a message."""
         raise NotImplementedError("Subclasses must implement this method.")
 
+    @abstractmethod
+    def create(self, *, queue: str):
+        """Create a queue if it doesn't exist."""
+        raise NotImplementedError("Subclasses must implement this method.")
+
     @abstractmethod
     def purge(self, *, queue: str):
         """Purge all messages from the queue."""

queueio/broker_test.py

@@ -60,6 +60,7 @@ def broker(self):
     @pytest.mark.timeout(2)
     def test_prefetch_limits_message_consumption(self, broker):
         """Verify that prefetch parameter limits message consumption."""
+        broker.create(queue="test-queue")
         broker.purge(queue="test-queue")
 
         # Enqueue more messages than prefetch limit
@@ -92,6 +93,7 @@ def receive_messages():
     @pytest.mark.timeout(2)
     def test_suspend_resume_affects_prefetch_capacity(self, broker):
         """Verify suspending messages frees capacity and resuming reduces it."""
+        broker.create(queue="test-queue")
         broker.purge(queue="test-queue")
 
         # Enqueue messages
@@ -129,6 +131,7 @@ def receive_messages():
     @pytest.mark.timeout(2)
     def test_complete_message_frees_prefetch_capacity(self, broker):
         """Verify completing messages frees up capacity."""
+        broker.create(queue="test-queue")
         broker.purge(queue="test-queue")
 
         # Enqueue more messages than prefetch limit
@@ -168,6 +171,7 @@ def receive_messages():
     @pytest.mark.timeout(2)
     def test_multiple_receivers_independent_prefetch_limits(self, broker):
         """Verify multiple receivers operate with independent prefetch limits."""
+        broker.create(queue="test-queue")
         broker.purge(queue="test-queue")
 
         # Enqueue enough messages for both receivers
@@ -219,6 +223,8 @@ def test_receive_rejects_empty_queues(self, broker):
     @skip_if_unsupported("supports_multiple_queues")
     def test_receive_supports_multiple_queues(self, broker):
         """Verify broker can receive from multiple queues."""
+        broker.create(queue="queue1")
+        broker.create(queue="queue2")
         broker.purge(queue="queue1")
         broker.purge(queue="queue2")
 
@@ -254,6 +260,10 @@ def receive_messages():
     def test_multiple_queues_with_mixed_empty_and_filled(self, broker):
         """Verify broker handles multiple queues with some empty, some with messages."""
 
+        broker.create(queue="empty1")
+        broker.create(queue="filled")
+        broker.create(queue="empty2")
+        broker.create(queue="also_filled")
         broker.purge(queue="empty1")
         broker.purge(queue="filled")
         broker.purge(queue="empty2")
@@ -298,6 +308,8 @@ def test_duplicate_queue_names_get_additional_priority(self, broker):
         probability in the round-robin selection.
         """
 
+        broker.create(queue="priority_queue")
+        broker.create(queue="normal_queue")
         broker.purge(queue="priority_queue")
         broker.purge(queue="normal_queue")
 
@@ -398,6 +410,9 @@ def test_empty_queue_cycling_fairness(self, broker):
         the message, not just cycle by a fixed amount.
         """
 
+        broker.create(queue="empty")
+        broker.create(queue="queue1")
+        broker.create(queue="queue2")
         broker.purge(queue="empty")
         broker.purge(queue="queue1")
         broker.purge(queue="queue2")

queueio/pika/broker.py

@@ -26,11 +26,12 @@ def __init__(self, connection_params: Parameters):
         self.__receivers = set[PikaReceiver]()
 
     def enqueue(self, body: bytes, /, *, queue: str):
-        self.__channel.declare_queue(queue=queue, durable=True)
         self.__channel.publish(exchange="", routing_key=queue, body=body)
 
-    def purge(self, *, queue: str):
+    def create(self, *, queue: str):
         self.__channel.declare_queue(queue=queue, durable=True)
+
+    def purge(self, *, queue: str):
         self.__channel.purge(queue=queue)
 
     def receive(self, queuespec: QueueSpec, /) -> PikaReceiver:

queueio/pika/receiver.py

@@ -27,11 +27,7 @@ def __init__(
         self.__prefetch = 0
         self.__adjust_prefetch(+queuespec.concurrency)
 
-        declared_queues = set()
         for queue in queuespec.queues:
-            if queue not in declared_queues:
-                self.__channel.declare_queue(queue=queue, durable=True)
-                declared_queues.add(queue)
             result = self.__channel.consume(queue)
             self.__consumer_tag[queue] = cast(str, result.method.consumer_tag)
 

queueio/queueio.py

@@ -112,6 +112,9 @@ def run[R](self, invocation: Invocation[R], /) -> R:
         with self.invocation_handler():
             return invocation.submit().result()
 
+    def create(self, *, queue: str):
+        self.__broker.create(queue=queue)
+
     def purge(self, *, queue: str):
         self.__broker.purge(queue=queue)
 

queueio/queueio_test.py

@@ -17,6 +17,7 @@ def test_queueio_with_custom_broker_and_journal():
 
     try:
         # Test purge (uses broker)
+        queueio.create(queue="queueio")
         queueio.purge(queue="queueio")
 
         # Test subscriptions (uses journal)
@@ -39,7 +40,9 @@ def test_different_queueio_instances_are_independent():
 
     try:
         # Both should work independently
+        queueio1.create(queue="queueio")
         queueio1.purge(queue="queueio")
+        queueio2.create(queue="queueio")
         queueio2.purge(queue="queueio")
 
         # Test that they can have independent subscriptions
@@ -83,6 +86,7 @@ def test_queueio_loads_configuration_from_pyproject(tmp_path):
         queueio = QueueIO()
         try:
             # Should be able to perform basic operations
+            queueio.create(queue="test")
             queueio.purge(queue="test")
             events = queueio.subscribe({object})
             queueio.unsubscribe(events)
@@ -132,7 +136,9 @@ def test_queueio_allows_pika_override(tmp_path):
 
         try:
             # Both should work
+            queueio1.create(queue="test")
             queueio1.purge(queue="test")
+            queueio2.create(queue="test")
             queueio2.purge(queue="test")
         finally:
             queueio1.shutdown()
@@ -203,6 +209,7 @@ def test_queueio_with_valid_config(tmp_path):
         queueio = QueueIO()
         try:
             # Should load configuration successfully
+            queueio.create(queue="test")
             queueio.purge(queue="test")
             routines = queueio.routines()
             routine_names = {routine.name for routine in routines}
@@ -320,6 +327,7 @@ def test_queueio_with_uri_broker_config(tmp_path):
         queueio = QueueIO()
         try:
             # Should work with URI configuration
+            queueio.create(queue="test")
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()
@@ -359,6 +367,7 @@ def test_queueio_with_uri_journal_config(tmp_path):
         queueio = QueueIO()
         try:
             # Should work with URI configuration
+            queueio.create(queue="test")
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()
@@ -398,6 +407,7 @@ def test_queueio_with_both_uri_configs(tmp_path):
         queueio = QueueIO()
         try:
             # Should work with both URI configurations
+            queueio.create(queue="test")
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()
@@ -478,6 +488,7 @@ def test_queueio_with_environment_variable(tmp_path, monkeypatch):
         queueio = QueueIO()
         try:
             # Should work with environment variables taking precedence
+            queueio.create(queue="test")
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()

queueio/samples/basic_test.py

@@ -14,13 +14,14 @@ def test_integration():
     queueio = QueueIO()
 
     with queueio.activate():
+        queueio.create(queue="queueio")
         queueio.purge(queue="queueio")
         events = queueio.subscribe({Invocation.Completed})
         invocation = yielding(7)
         queueio.submit(invocation)
 
         proc = subprocess.Popen(
-            [sys.executable, "-m", "queueio", "worker", "queueio=1"],
+            [sys.executable, "-m", "queueio", "run", "queueio=1"],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
         )

queueio/samples/expanded_test.py

@@ -15,14 +15,15 @@ def test_integration():
     queueio = QueueIO()
 
     with queueio.activate():
+        queueio.create(queue="queueio")
         queueio.purge(queue="queueio")
         events = queueio.subscribe({Invocation.Completed})
         invocation = irregular()
         queueio.submit(invocation)
 
         # 1. Start worker process in the background
         proc = subprocess.Popen(
-            [sys.executable, "-m", "queueio", "worker", "queueio=1"],
+            [sys.executable, "-m", "queueio", "run", "queueio=1"],
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
         )