Manage broker and journal lifecycles

Introduce a Backend ABC that can make a shared connection for broker and
journal.

Author: ryanhiebert

CHANGELOG.md

@@ -8,8 +8,13 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com).
 [Unreleased]
 ------------
 
+[0.7.0] - 2026-03-09
+--------------------
+
 ### Breaking Changes
 
+- The `pika` configuration key in `[tool.queueio]` has been renamed to `broker`.
+- The `QUEUEIO_PIKA` environment variable has been renamed to `QUEUEIO_BROKER`.
 - 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])`.
@@ -19,6 +24,8 @@ The format is loosely based on [Keep a Changelog](https://keepachangelog.com).
 
 - Added `Broker.sync()` to back `queueio sync`.
 - Added `QueueIO.sync()` to back `queueio sync`.
+- `Backend` ABC with `broker()` and `journal()` methods
+  that return independent context managers with their own lifecycles.
 
 [0.6.0] - 2026-02-16
 --------------------
@@ -118,7 +125,8 @@ Thank you to Nick Anderegg for allowing me to use the queueio name for this proj
 - The queuespec syntax to `queue run` to consume multiple queues with shared capacity.
 - `queueio monitor` command to monitor activity in the queueio system.
 
-[Unreleased]: https://github.com/ryanhiebert/queueio/compare/tag/0.6.0...HEAD
+[Unreleased]: https://github.com/ryanhiebert/queueio/compare/tag/0.7.0...HEAD
+[0.7.0]: https://github.com/ryanhiebert/queueio/compare/tag/0.6.0...tag/0.7.0
 [0.6.0]: https://github.com/ryanhiebert/queueio/compare/tag/0.5.0...tag/0.6.0
 [0.5.0]: https://github.com/ryanhiebert/queueio/compare/tag/0.4.0...tag/0.5.0
 [0.4.0]: https://github.com/ryanhiebert/queueio/compare/tag/0.3.0...tag/0.4.0

README.md

@@ -46,17 +46,17 @@ Add the configuration to your `pyproject.toml`:
 
 ```toml
 [tool.queueio]
-# Configure RabbitMQ
-pika = "amqp://guest:guest@localhost:5672/"
+# Configure the broker (currently supports amqp:// URIs)
+broker = "amqp://guest:guest@localhost:5672/"
 # Register the modules that the worker should load to find your routines
 register = ["basic"]
 ```
 
-The pika configuration can be overridden with an environment variable
+The broker configuration can be overridden with an environment variable
 to allow a project to be deployed in multiple environments.
 
 ```sh
-QUEUEIO_PIKA='amqp://guest:guest@localhost:5672/'
+QUEUEIO_BROKER='amqp://guest:guest@localhost:5672/'
 ```
 
 Sync the queues to the broker:

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "queueio"
-version = "0.6.0"
+version = "0.7.0"
 description = "Python background queues with an async twist"
 readme = "README.md"
 license = "MIT"
@@ -68,7 +68,7 @@ register = [
     "queueio.samples.priority",
     "queueio.samples.queuevar",
 ]
-pika = "amqp://localhost:5672"
+broker = "amqp://localhost:5672"
 
 [tool.pytest.ini_options]
 timeout = 30

queueio/__init__.py

@@ -2,13 +2,13 @@
 
 from .gather import gather as gather
 from .pause import pause as pause
-from .queueio import QueueIO as RealQueueIO
+from .queueio import QueueIO as QueueIO
 from .queueio import priority as priority
 from .queuevar import QueueVar as QueueVar
 from .registry import routine as routine
 
 
 @contextmanager
 def activate():
-    with RealQueueIO().activate():
+    with QueueIO.default() as queueio, queueio.activate():
         yield

queueio/__main__.py

@@ -31,8 +31,7 @@
 @routine_app.command("list")
 def routine_list():
     """Show all registered routines."""
-    queueio = QueueIO()
-    try:
+    with QueueIO.default() as queueio:
         routines = queueio.routines()
 
         if not routines:
@@ -52,8 +51,6 @@ def routine_list():
         print(f"{'-' * name_width}-+-{'-' * path_width}")
         for routine, path in zip(routines, function_paths, strict=False):
             print(f"{routine.name:<{name_width}} | {path:<{path_width}}")
-    finally:
-        queueio.shutdown()
 
 
 @app.command(rich_help_panel="Commands")
@@ -63,17 +60,16 @@ def monitor(raw: bool = False):
     Show a live view of queueio activity. Use --raw for detailed event output.
     """
     if raw:
-        queueio = QueueIO()
-        events = queueio.subscribe({object})
-        try:
-            while True:
-                print(events.get())
-        except KeyboardInterrupt:
-            print("Shutting down gracefully.")
-        finally:
-            queueio.shutdown()
+        with QueueIO.default() as queueio:
+            events = queueio.subscribe({object})
+            try:
+                while True:
+                    print(events.get())
+            except KeyboardInterrupt:
+                print("Shutting down gracefully.")
     else:
-        Monitor().run()
+        with QueueIO.default() as queueio:
+            Monitor(queueio).run()
 
 
 @app.command(rich_help_panel="Commands")
@@ -93,8 +89,8 @@ def run(
     The worker will process invocations from the specified queue,
     as many at a time as specified by the concurrency.
     """
-    queueio = QueueIO()
-    Worker(queueio, queuespec)()
+    with QueueIO.default() as queueio:
+        Worker(queueio, queuespec)()
 
 
 @app.command(rich_help_panel="Commands")
@@ -108,8 +104,7 @@ def sync(
     ] = False,
 ):
     """Sync resources for the broker and journal."""
-    queueio = QueueIO()
-    try:
+    with QueueIO.default() as queueio:
         routines = queueio.routines()
 
         if not routines:
@@ -133,8 +128,6 @@ def sync(
             raise typer.Exit(1) from None
 
         print(f"Successfully synced {len(queues)} queue(s)")
-    finally:
-        queueio.shutdown()
 
 
 @queue_app.command("purge")
@@ -153,8 +146,7 @@ def queue_purge(
     This will remove all pending messages from the given queues.
     Use with caution as this operation cannot be undone.
     """
-    queueio = QueueIO()
-    try:
+    with QueueIO.default() as queueio:
         queue_list = [q.strip() for q in queues.split(",") if q.strip()]
         if not queue_list:
             print("Error: No valid queue names provided")
@@ -165,8 +157,6 @@ def queue_purge(
             queueio.purge(queue=queue)
 
         print(f"Successfully purged {len(queue_list)} queue(s)")
-    finally:
-        queueio.shutdown()
 
 
 if __name__ == "__main__":

queueio/backend.py

@@ -0,0 +1,18 @@
+from abc import ABC
+from abc import abstractmethod
+from contextlib import AbstractContextManager
+
+from .broker import Broker
+from .journal import Journal
+
+
+class Backend(ABC):
+    """A backend provides context-managed access to a broker and journal."""
+
+    @abstractmethod
+    def broker(self) -> AbstractContextManager[Broker]:
+        raise NotImplementedError("Subclasses must implement this method.")
+
+    @abstractmethod
+    def journal(self) -> AbstractContextManager[Journal]:
+        raise NotImplementedError("Subclasses must implement this method.")

queueio/broker.py

@@ -14,12 +14,6 @@ class Broker(ABC):
     at-least-once delivery.
     """
 
-    @classmethod
-    @abstractmethod
-    def from_uri(cls, uri: str, /):
-        """Create a broker instance from a URI."""
-        raise NotImplementedError("Subclasses must implement this method.")
-
     @abstractmethod
     def sync(self, queues: Iterable[str], *, recreate: bool = False):
         """Ensure the given queues are ready to use.

queueio/journal.py

@@ -4,12 +4,6 @@
 
 
 class Journal(ABC):
-    @classmethod
-    @abstractmethod
-    def from_uri(cls, uri: str, /):
-        """Create a journal instance from a URI."""
-        raise NotImplementedError("Subclasses must implement this method.")
-
     @abstractmethod
     def subscribe(self) -> Iterator[bytes]:
         raise NotImplementedError("Subclasses must implement this method.")

queueio/monitor.py

@@ -17,9 +17,9 @@ class Monitor(App):
 
     TITLE = "queueio Monitor"
 
-    def __init__(self):
+    def __init__(self, queueio: QueueIO):
         super().__init__()
-        self.__queueio = QueueIO()
+        self.__queueio = queueio
         self.__thread = Thread(target=self.__listen)
         self.__events = self.__queueio.subscribe(
             {

queueio/pika/__init__.py

@@ -0,0 +1,31 @@
+from collections.abc import Generator
+from contextlib import AbstractContextManager
+from contextlib import contextmanager
+
+from pika import URLParameters
+
+from queueio.backend import Backend
+
+from .broker import PikaBroker
+from .journal import PikaJournal
+from .threadsafe import ThreadsafeConnection
+
+
+class PikaBackend(Backend):
+    def __init__(self, connection: ThreadsafeConnection):
+        self.__connection = connection
+
+    @classmethod
+    @contextmanager
+    def connect(cls, uri: str, /) -> Generator[PikaBackend]:
+        connection = ThreadsafeConnection(URLParameters(uri))
+        try:
+            yield cls(connection)
+        finally:
+            connection.close()
+
+    def broker(self) -> AbstractContextManager[PikaBroker]:
+        return PikaBroker.connect(self.__connection)
+
+    def journal(self) -> AbstractContextManager[PikaJournal]:
+        return PikaJournal.connect(self.__connection)