Require a default queue

Author: ryanhiebert

qio/__main__.py

@@ -23,7 +23,7 @@ def regular(instance: int, iterations: int):
     return f"Instance {instance} completed"
 
 
-@routine(name="raises", queue="qio")
+@routine(name="raises")
 def raises():
     raise ValueError("This is a test exception")
 
@@ -60,7 +60,9 @@ async def irregular():
 def submit():
     connection_params = ConnectionParameters()
     qio = Qio(
-        broker=PikaBroker(connection_params), transport=PikaTransport(connection_params)
+        broker=PikaBroker(connection_params),
+        transport=PikaTransport(connection_params),
+        default_queue="qio",
     )
     try:
         qio.submit(irregular())
@@ -75,6 +77,7 @@ def monitor(raw: bool = False):
         qio = Qio(
             broker=PikaBroker(connection_params),
             transport=PikaTransport(connection_params),
+            default_queue="qio",
         )
         events = qio.subscribe({object})
         try:
@@ -92,7 +95,9 @@ def monitor(raw: bool = False):
 def worker():
     connection_params = ConnectionParameters()
     qio = Qio(
-        broker=PikaBroker(connection_params), transport=PikaTransport(connection_params)
+        broker=PikaBroker(connection_params),
+        transport=PikaTransport(connection_params),
+        default_queue="qio",
     )
     Worker(qio, queue="qio", concurrency=3)()
 
@@ -101,7 +106,9 @@ def worker():
 def purge():
     connection_params = ConnectionParameters()
     qio = Qio(
-        broker=PikaBroker(connection_params), transport=PikaTransport(connection_params)
+        broker=PikaBroker(connection_params),
+        transport=PikaTransport(connection_params),
+        default_queue="qio",
     )
     try:
         qio.purge(queue="qio")

qio/monitor.py

@@ -32,6 +32,7 @@ def __init__(self):
         self.__qio = Qio(
             broker=PikaBroker(connection_params),
             transport=PikaTransport(connection_params),
+            default_queue="qio",
         )
         self.__thread = Thread(target=self.__listen)
         self.__events = self.__qio.subscribe(

qio/qio.py

@@ -31,9 +31,12 @@
 
 
 class Qio:
-    def __init__(self, *, broker: Broker, transport: Transport):
+    def __init__(
+        self, *, broker: Broker, transport: Transport, default_queue: str = "default"
+    ):
         self.__bus = Bus(transport)
         self.__broker = broker
+        self.__default_queue = default_queue
         self.__invocations = dict[Invocation, Message]()
 
     def run[R](self, invocation: Invocation[R], /) -> R:
@@ -103,7 +106,8 @@ def submit(self, invocation: Invocation, /):
                 kwargs=invocation.kwargs,
             )
         )
-        self.__broker.enqueue(serialize(invocation), queue=routine.queue)
+        queue = routine.queue if routine.queue is not None else self.__default_queue
+        self.__broker.enqueue(serialize(invocation), queue=queue)
 
     def consume(self, *, queue: str, prefetch: int) -> Generator[Invocation]:
         for message in self.__broker.consume(queue=queue, prefetch=prefetch):

qio/qio_test.py

@@ -7,7 +7,7 @@ def test_qio_with_custom_broker_and_transport():
     """Qio custom broker and transport implementations."""
     broker = StubBroker()
     transport = StubTransport()
-    qio = Qio(broker=broker, transport=transport)
+    qio = Qio(broker=broker, transport=transport, default_queue="qio")
 
     try:
         # Test purge (uses broker)
@@ -26,11 +26,11 @@ def test_different_qio_instances_are_independent():
     # Create two Qio instances with different stub implementations
     broker1 = StubBroker()
     transport1 = StubTransport()
-    qio1 = Qio(broker=broker1, transport=transport1)
+    qio1 = Qio(broker=broker1, transport=transport1, default_queue="qio")
 
     broker2 = StubBroker()
     transport2 = StubTransport()
-    qio2 = Qio(broker=broker2, transport=transport2)
+    qio2 = Qio(broker=broker2, transport=transport2, default_queue="qio")
 
     try:
         # Both should work independently

qio/registry.py

@@ -5,7 +5,7 @@
 ROUTINE_REGISTRY: dict[str, Routine] = {}
 
 
-def routine(*, name: str, queue: str):
+def routine(*, name: str, queue: str | None = None):
     """Decorate a function to make it a routine."""
 
     def create_routine[**A, R](fn: Callable[A, R]) -> Routine[A, R]:

qio/routine.py

@@ -4,7 +4,7 @@
 
 
 class Routine[**A, R]:
-    def __init__(self, fn: Callable[A, R], *, name: str, queue: str):
+    def __init__(self, fn: Callable[A, R], *, name: str, queue: str | None = None):
         self.fn = fn
         self.name = name
         self.queue = queue

qio_test.py

@@ -16,7 +16,9 @@ def test_integration():
     # Prefers a clean environment and queue
     connection_params = ConnectionParameters()
     qio = Qio(
-        broker=PikaBroker(connection_params), transport=PikaTransport(connection_params)
+        broker=PikaBroker(connection_params),
+        transport=PikaTransport(connection_params),
+        default_queue="qio",
     )
 
     try: