Release v0.2.160

Queues support batching on consume

Author: andrevanzuydam

README.md

@@ -87,6 +87,10 @@ https://tina4.com/
 MIT © 2007 – 2025 Tina4 Stack  
 https://opensource.org/licenses/MIT
 
+## Testing 
+
+uv run pytest --verbose
+
 ---
 
 **Tina4** – The framework that keeps out of the way of your coding.

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "tina4-python"
-version = "0.2.159"
+version = "0.2.160"
 description = "Tina4Python - This is not another framework for Python"
 authors = [
     {name = "Andre van Zuydam",email = "andrevanzuydam@gmail.com"}

tests/test_mongo_queue.py

@@ -112,6 +112,50 @@ def test_consumer_wrapper(mongo_config):
     assert len(collected) == 1
     assert collected[0].data == "to consume mongo"
 
+def test_batch_mode():
+    callback_calls = []
+
+    def my_callback(msg: Message):
+        callback_calls.append(msg.data)
+
+    # Create queue with batch_size=5 and callback
+    queue = Queue(topic="batch-test", batch_size=5, callback=my_callback)
+
+    # Produce 12 messages
+    for i in range(12):
+        queue.produce(f"Message {i+1}", user_id="tester")
+        time.sleep(0.2)
+
+    # Consume in batch mode
+    batches = []
+    total_messages = 0
+
+    consumer = Consumer([queue], acknowledge=True)
+
+    for item in consumer.messages():
+        assert isinstance(item, list)
+        assert all(isinstance(m, Message) for m in item)
+        batch_size = len(item)
+        batches.append(batch_size)
+        total_messages += batch_size
+
+
+        # Extract data for easier checking
+        batch_data = [msg.data for msg in item]
+        print("BATCH_DATA", batch_data)
+        expected_data = [f"Message {total_messages - batch_size + j + 1}" for j in range(batch_size)]
+        assert batch_data == expected_data
+
+        if total_messages >= 12:
+            break
+
+    # Verify we got the expected batch sizes: 5 + 5 + 2
+    assert batches == [5, 5, 2]
+    assert total_messages == 12
+    assert len(callback_calls) == 12  # Callback called once per message
+    assert sorted(callback_calls) == sorted([f"Message {i+1}" for i in range(12)])
+
+
 def test_error_handling(mongo_config):
     topic = unique_topic("error")
     clear_mongo_channel(topic, prefix=mongo_config.prefix)

tests/test_queues.py

@@ -3,6 +3,7 @@
 import pytest
 from tina4_python.Queue import Queue, Config, Message, Producer, Consumer
 
+
 # Helper to clean up SQLite files completely
 def cleanup_db(db_name):
     for suffix in ["", "-wal", "-shm", "-journal"]:
@@ -17,6 +18,7 @@ def cleanup_db(db_name):
                 except:
                     pass  # best effort
 
+
 @pytest.fixture(scope="function")
 def litequeue_config():
     # Unique DB name per test run to avoid locks
@@ -28,17 +30,20 @@ def litequeue_config():
     yield config
     cleanup_db(config.litequeue_database_name)
 
+
 @pytest.fixture(scope="function")
 def lite_queue(litequeue_config):
     q = Queue(config=litequeue_config, topic="test-topic")
     yield q
     cleanup_db(litequeue_config.litequeue_database_name)
 
+
 def test_init_litequeue(lite_queue):
     assert lite_queue.producer is not None
     assert lite_queue.consumer is not None
     assert lite_queue.get_prefix() == "test_"
 
+
 def test_produce_litequeue(lite_queue):
     response = lite_queue.produce("hello world", user_id="user123")
     assert isinstance(response, Message)
@@ -47,6 +52,7 @@ def test_produce_litequeue(lite_queue):
     assert response.status == 0
     assert len(response.message_id) == 36
 
+
 def test_produce_with_callback(lite_queue):
     delivered = [None]
 
@@ -58,6 +64,7 @@ def delivery_cb(producer, err, msg):
     assert isinstance(delivered[0][1], Message)
     assert delivered[0][1].data == "callback test"
 
+
 def test_consume_litequeue(lite_queue):
     lite_queue.produce("consume me", user_id="user789")
 
@@ -69,6 +76,7 @@ def test_consume_litequeue(lite_queue):
     assert msg.data == "consume me"
     assert msg.status == 2  # Acknowledged
 
+
 def test_consume_no_ack(lite_queue):
     lite_queue.produce("no ack test", user_id="user000")
 
@@ -77,13 +85,15 @@ def test_consume_no_ack(lite_queue):
     msg = messages[0]
     assert msg.status == 1  # Not acknowledged
 
+
 def test_producer_wrapper(litequeue_config):
     q = Queue(config=litequeue_config, topic="producer-test")
     producer = Producer(q)
     response = producer.produce("wrapped produce", user_id="wrapped_user")
     assert isinstance(response, Message)
     assert response.data == "wrapped produce"
 
+
 def test_consumer_wrapper(litequeue_config):
     q = Queue(config=litequeue_config, topic="consumer-test")
     q.produce("to consume", user_id="consume_user")
@@ -100,10 +110,55 @@ def test_consumer_wrapper(litequeue_config):
     assert len(collected) >= 1
     assert any(m.data == "to consume" for m in collected)
 
+
+def test_batch_mode():
+    callback_calls = []
+
+    def my_callback(msg: Message):
+        callback_calls.append(msg.data)
+
+    # Create queue with batch_size=5 and callback
+    queue = Queue(topic="batch-test", batch_size=5, callback=my_callback)
+
+    # Produce 12 messages
+    for i in range(12):
+        queue.produce(f"Message {i + 1}", user_id="tester")
+        time.sleep(0.1)
+
+    # Consume in batch mode
+    batches = []
+    total_messages = 0
+
+    consumer = Consumer([queue], acknowledge=True)
+
+    for item in consumer.messages():
+        assert isinstance(item, list)
+        assert all(isinstance(m, Message) for m in item)
+        batch_size = len(item)
+        batches.append(batch_size)
+        total_messages += batch_size
+
+        # Extract data for easier checking
+        batch_data = [msg.data for msg in item]
+
+        expected_data = [f"Message {total_messages - batch_size + j + 1}" for j in range(batch_size)]
+        print("BATCH DATA", batch_data, expected_data)
+        assert batch_data == expected_data
+
+        if total_messages >= 12:
+            break
+
+    # Verify we got the expected batch sizes: 5 + 5 + 2
+    assert batches == [5, 5, 2]
+    assert total_messages == 12
+    assert len(callback_calls) == 12  # Callback called once per message
+    assert sorted(callback_calls) == sorted([f"Message {i + 1}" for i in range(12)])
+
+
 def test_error_handling(lite_queue):
     with pytest.raises(Exception):
         lite_queue.produce(None)
 
     # Consume on empty queue — should yield nothing, no error
     messages = list(lite_queue.consume())
-    assert len(messages) == 0
+    assert len(messages) == 0

tina4_python/Queue.py

@@ -55,14 +55,15 @@ class Message:
     delivery_tag: str
 
 class Queue:
-    def __init__(self, config=None, topic="default-queue", callback=None):
+    def __init__(self, config=None, topic="default-queue", callback=None, batch_size=1):
         if config is None:
             config = Config()
         self.config = config
         self.topic = topic
         self.callback = callback
         self.producer = None
         self.consumer = None
+        self.batch_size = batch_size
         init_method = f"init_{config.queue_type.replace('-', '_')}"
         getattr(self, init_method)()
 
@@ -101,7 +102,7 @@ def kafka_cb(err, kafka_msg):
                 delivery_callback(self.producer, e, None)
             return e
 
-    def consume(self, acknowledge: bool = True) -> Generator[Message, None, None]:
+    def consume(self, acknowledge: bool = True) -> Generator[Message | List[Message], None, None]:
         """
         Generator that continuously yields messages from the queue as they arrive.
         Use like:
@@ -110,9 +111,12 @@ def consume(self, acknowledge: bool = True) -> Generator[Message, None, None]:
         If a callback was provided in __init__, it will also be called for each message.
         """
         prefix = self.get_prefix()
+        is_batch = self.batch_size > 1
+        count_messages = 0
         try:
             message_found = True
-            while message_found:
+            batch = []
+            while message_found and count_messages < self.batch_size:
                 response = None
                 message_found = False
 
@@ -153,8 +157,9 @@ def consume(self, acknowledge: bool = True) -> Generator[Message, None, None]:
                         if acknowledge:
                             self.consumer.commit()
 
-                if response is not None:
-                    yield response
+                if message_found:
+                    count_messages += 1
+                    batch.append(response)
                     if self.callback:
                         try:
                             self.callback(response)
@@ -164,6 +169,11 @@ def consume(self, acknowledge: bool = True) -> Generator[Message, None, None]:
                     # No message available right now — brief sleep to avoid busy loop
                     time.sleep(0.05)
 
+            if len(batch) > 0:
+                if not is_batch:
+                    yield batch[0]
+                else:
+                    yield batch
         except Exception as e:
             Debug.error(f"Error consuming {self.topic}: {e}")
             raise  # Re-raise to stop consumption on fatal error