Add Broker sync method

Replace per-queue create()/delete() with sync(queues) that takes all queues at once. This better fits backends like PostgreSQL where setup means creating tables rather than declaring individual queues.

The CLI sync command now delegates directly to broker.sync() instead of looping over queues itself. The --recreate flag is passed through so each backend handles recreation internally.

Author: ryanhiebert

CHANGELOG.md

@@ -8,6 +8,18 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com).
 [Unreleased]
 ------------
 
+### Breaking Changes
+
+- Removed `Broker.create(queue)`, in favor of `Broker.sync([queue])`.
+- Removed `Broker.delete(queue)`, without replacement.
+- Removed `QueueIO.create(queue)`, in favor of `QueueIO.sync([queue])`.
+- Removed `QueueIO.delete(queue)`, without replacement.
+
+### Added
+
+- Added `Broker.sync()` to back `queueio sync`.
+- Added `QueueIO.sync()` to back `queueio sync`.
+
 [0.6.0] - 2026-02-16
 --------------------
 

queueio/__main__.py

@@ -1,4 +1,3 @@
-from contextlib import suppress
 from typing import Annotated
 
 import typer
@@ -103,20 +102,12 @@ def sync(
     recreate: Annotated[
         bool,
         typer.Option(
-            help="Delete and recreate queues that have incompatible arguments. "
-            "WARNING: This will lose any pending messages in those queues.",
+            help="Recreate all resources for the broker and journal. "
+            "WARNING: This will lose pending messages.",
         ),
     ] = False,
 ):
-    """Sync known queues to the broker.
-
-    Planned flags:
-    - (default) ensure everything needed exists
-    - --repair: rebuild only what's broken
-    - --recreate: tear down and rebuild everything we know about
-    - --prune: remove what's not needed
-    - --plan: show what would happen without doing it
-    """
+    """Sync resources for the broker and journal."""
     queueio = QueueIO()
     try:
         routines = queueio.routines()
@@ -127,29 +118,19 @@ def sync(
 
         queues = sorted({routine.queue for routine in routines})
 
-        print(f"Syncing queues for {len(routines)} routine(s):")
-        if recreate:
-            for queue in queues:
-                print(f"  Recreating queue: {queue}")
-                with suppress(Exception):
-                    queueio.delete(queue=queue)
-
-        failed = []
+        print(f"Discovered queues for {len(routines)} routine(s):")
         for queue in queues:
-            print(f"  Ensuring queue exists: {queue}")
-            try:
-                queueio.create(queue=queue)
-            except Exception:
-                failed.append(queue)
+            print(f"  {queue}")
 
-        if failed:
+        try:
+            queueio.sync(queues, recreate=recreate)
+        except Exception:
             print(
-                f"\nError: {len(failed)} queue(s) have incompatible arguments: "
-                f"{', '.join(failed)}\n"
-                f"Re-run with --recreate to delete and recreate them.\n"
-                f"WARNING: This will lose any pending messages in those queues."
+                "\nError: Failed to sync resources.\n"
+                "Re-run with --recreate to recreate them.\n"
+                "WARNING: This will lose pending messages."
             )
-            raise typer.Exit(1)
+            raise typer.Exit(1) from None
 
         print(f"Successfully synced {len(queues)} queue(s)")
     finally:

queueio/broker.py

@@ -1,5 +1,6 @@
 from abc import ABC
 from abc import abstractmethod
+from collections.abc import Iterable
 
 from .queuespec import QueueSpec
 from .receiver import Receiver
@@ -20,18 +21,17 @@ def from_uri(cls, uri: str, /):
         raise NotImplementedError("Subclasses must implement this method.")
 
     @abstractmethod
-    def enqueue(self, body: bytes, /, *, queue: str, priority: int):
-        """Enqueue a message."""
-        raise NotImplementedError("Subclasses must implement this method.")
+    def sync(self, queues: Iterable[str], *, recreate: bool = False):
+        """Ensure the given queues are ready to use.
 
-    @abstractmethod
-    def create(self, *, queue: str):
-        """Create a queue if it doesn't exist."""
+        If recreate is True, destroy and recreate all resources,
+        losing any pending messages.
+        """
         raise NotImplementedError("Subclasses must implement this method.")
 
     @abstractmethod
-    def delete(self, *, queue: str):
-        """Delete a queue."""
+    def enqueue(self, body: bytes, /, *, queue: str, priority: int):
+        """Enqueue a message."""
         raise NotImplementedError("Subclasses must implement this method.")
 
     @abstractmethod

queueio/broker_test.py

@@ -60,7 +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.sync(["test-queue"])
         broker.purge(queue="test-queue")
 
         # Enqueue more messages than prefetch limit
@@ -93,7 +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.sync(["test-queue"])
         broker.purge(queue="test-queue")
 
         # Enqueue messages
@@ -131,7 +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.sync(["test-queue"])
         broker.purge(queue="test-queue")
 
         # Enqueue more messages than prefetch limit
@@ -171,7 +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.sync(["test-queue"])
         broker.purge(queue="test-queue")
 
         # Enqueue enough messages for both receivers
@@ -223,8 +223,7 @@ 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.sync(["queue1", "queue2"])
         broker.purge(queue="queue1")
         broker.purge(queue="queue2")
 
@@ -260,10 +259,7 @@ 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.sync(["empty1", "filled", "empty2", "also_filled"])
         broker.purge(queue="empty1")
         broker.purge(queue="filled")
         broker.purge(queue="empty2")
@@ -308,8 +304,7 @@ 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.sync(["priority_queue", "normal_queue"])
         broker.purge(queue="priority_queue")
         broker.purge(queue="normal_queue")
 
@@ -410,9 +405,7 @@ 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.sync(["empty", "queue1", "queue2"])
         broker.purge(queue="empty")
         broker.purge(queue="queue1")
         broker.purge(queue="queue2")
@@ -473,7 +466,7 @@ def receive_messages():
     @pytest.mark.timeout(2)
     def test_higher_priority_consumed_first(self, broker):
         """Verify that higher priority messages are consumed before lower ones."""
-        broker.create(queue="test-queue")
+        broker.sync(["test-queue"])
         broker.purge(queue="test-queue")
 
         # Enqueue low priority first, then high priority

queueio/pika/broker.py

@@ -1,3 +1,5 @@
+from collections.abc import Iterable
+from contextlib import suppress
 from threading import Lock
 
 from pika import URLParameters
@@ -26,6 +28,19 @@ def __init__(self, connection_params: Parameters):
         self.__shutdown = False
         self.__receivers = set[PikaReceiver]()
 
+    def sync(self, queues: Iterable[str], *, recreate: bool = False):
+        channel = self.__connection.channel()
+        try:
+            for queue in queues:
+                if recreate:
+                    with suppress(Exception):
+                        channel.delete(queue=queue)
+                channel.queue_declare(
+                    queue=queue, durable=True, arguments={"x-max-priority": 9}
+                )
+        finally:
+            channel.close()
+
     def enqueue(self, body: bytes, /, *, queue: str, priority: int):
         self.__channel.publish(
             exchange="",
@@ -34,18 +49,6 @@ def enqueue(self, body: bytes, /, *, queue: str, priority: int):
             properties=BasicProperties(priority=priority),
         )
 
-    def create(self, *, queue: str):
-        channel = self.__connection.channel()
-        channel.queue_declare(
-            queue=queue, durable=True, arguments={"x-max-priority": 9}
-        )
-        channel.close()
-
-    def delete(self, *, queue: str):
-        channel = self.__connection.channel()
-        channel.delete(queue=queue)
-        channel.close()
-
     def purge(self, *, queue: str):
         self.__channel.purge(queue=queue)
 

queueio/queueio.py

@@ -119,11 +119,8 @@ 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 delete(self, *, queue: str):
-        self.__broker.delete(queue=queue)
+    def sync(self, queues: Iterable[str], *, recreate: bool = False):
+        self.__broker.sync(queues, recreate=recreate)
 
     def purge(self, *, queue: str):
         self.__broker.purge(queue=queue)

queueio/queueio_test.py

@@ -17,7 +17,7 @@ def test_queueio_with_custom_broker_and_journal():
 
     try:
         # Test purge (uses broker)
-        queueio.create(queue="queueio")
+        queueio.sync(["queueio"])
         queueio.purge(queue="queueio")
 
         # Test subscriptions (uses journal)
@@ -40,9 +40,9 @@ def test_different_queueio_instances_are_independent():
 
     try:
         # Both should work independently
-        queueio1.create(queue="queueio")
+        queueio1.sync(["queueio"])
         queueio1.purge(queue="queueio")
-        queueio2.create(queue="queueio")
+        queueio2.sync(["queueio"])
         queueio2.purge(queue="queueio")
 
         # Test that they can have independent subscriptions
@@ -86,7 +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.sync(["test"])
             queueio.purge(queue="test")
             events = queueio.subscribe({object})
             queueio.unsubscribe(events)
@@ -136,9 +136,9 @@ def test_queueio_allows_pika_override(tmp_path):
 
         try:
             # Both should work
-            queueio1.create(queue="test")
+            queueio1.sync(["test"])
             queueio1.purge(queue="test")
-            queueio2.create(queue="test")
+            queueio2.sync(["test"])
             queueio2.purge(queue="test")
         finally:
             queueio1.shutdown()
@@ -209,7 +209,7 @@ def test_queueio_with_valid_config(tmp_path):
         queueio = QueueIO()
         try:
             # Should load configuration successfully
-            queueio.create(queue="test")
+            queueio.sync(["test"])
             queueio.purge(queue="test")
             routines = queueio.routines()
             routine_names = {routine.name for routine in routines}
@@ -327,7 +327,7 @@ def test_queueio_with_uri_broker_config(tmp_path):
         queueio = QueueIO()
         try:
             # Should work with URI configuration
-            queueio.create(queue="test")
+            queueio.sync(["test"])
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()
@@ -367,7 +367,7 @@ def test_queueio_with_uri_journal_config(tmp_path):
         queueio = QueueIO()
         try:
             # Should work with URI configuration
-            queueio.create(queue="test")
+            queueio.sync(["test"])
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()
@@ -407,7 +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.sync(["test"])
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()
@@ -488,7 +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.sync(["test"])
             queueio.purge(queue="test")
         finally:
             queueio.shutdown()

queueio/samples/basic_test.py

@@ -15,7 +15,7 @@ def test_integration():
     queueio = QueueIO()
 
     with queueio.activate():
-        queueio.create(queue="basic")
+        queueio.sync(["basic"])
         queueio.purge(queue="basic")
         events = queueio.subscribe({Invocation.Completed})
         invocation = yielding(7)

queueio/samples/expanded_test.py

@@ -16,7 +16,7 @@ def test_integration():
     queueio = QueueIO()
 
     with queueio.activate():
-        queueio.create(queue="expanded")
+        queueio.sync(["expanded"])
         queueio.purge(queue="expanded")
         events = queueio.subscribe({Invocation.Completed})
         invocation = irregular()

queueio/samples/priority_test.py

@@ -16,7 +16,7 @@ def test_priority_propagation():
     queueio = QueueIO()
 
     with queueio.activate():
-        queueio.create(queue="priority")
+        queueio.sync(["priority"])
         queueio.purge(queue="priority")
         events = queueio.subscribe({Invocation.Completed})
         invocation = demonstrate_priorities()