Repository: getlogbook/logbook (python)

Issue:
ThreadedWrapperHandler does not forward batched emits as expected
I noticed that the `ThreadedWrapperHandler` does not forward batch emits in *batch form*. Thus, if a `ThreadedWrapperHandler` wraps a `MailHandler`, a mail is sent for each logging entry instead of one summarising mail.

I may be able to provide a PR but let me ask first: Is this something you would like to get addressed or is the current behaviour considered *working as intended*?

---

Also, many thanks for this logging package! I use it for pretty much everything I write in Python. :+1:


Discussion:
@lgrahl Thanks for the feedback!

I think this is a reasonable change to implement. At the most naive approach this will be equivalent to a for loop with `emit`, and at the more complex scenarios this will enable all sorts of optimizations. Sounds like a good idea!

Patch:
diff --git a/logbook/queues.py b/logbook/queues.py
--- a/logbook/queues.py
+++ b/logbook/queues.py
@@ -604,7 +604,10 @@ class TWHThreadController(object):
     queue and sends it to a handler.  Both queue and handler are
     taken from the passed :class:`ThreadedWrapperHandler`.
     """
-    _sentinel = object()
+    class Command(object):
+        stop = object()
+        emit = object()
+        emit_batch = object()
 
     def __init__(self, wrapper_handler):
         self.wrapper_handler = wrapper_handler
@@ -621,17 +624,23 @@ def start(self):
     def stop(self):
         """Stops the task thread."""
         if self.running:
-            self.wrapper_handler.queue.put_nowait(self._sentinel)
+            self.wrapper_handler.queue.put_nowait((self.Command.stop, ))
             self._thread.join()
             self._thread = None
 
     def _target(self):
         while 1:
-            record = self.wrapper_handler.queue.get()
-            if record is self._sentinel:
+            item = self.wrapper_handler.queue.get()
+            command, data = item[0], item[1:]
+            if command is self.Command.stop:
                 self.running = False
                 break
-            self.wrapper_handler.handler.handle(record)
+            elif command is self.Command.emit:
+                (record, ) = data
+                self.wrapper_handler.handler.emit(record)
+            elif command is self.Command.emit_batch:
+                record, reason = data
+                self.wrapper_handler.handler.emit_batch(record, reason)
 
 
 class ThreadedWrapperHandler(WrapperHandler):
@@ -663,8 +672,17 @@ def close(self):
         self.handler.close()
 
     def emit(self, record):
+        item = (TWHThreadController.Command.emit, record)
         try:
-            self.queue.put_nowait(record)
+            self.queue.put_nowait(item)
+        except Full:
+            # silently drop
+            pass
+
+    def emit_batch(self, records, reason):
+        item = (TWHThreadController.Command.emit_batch, records, reason)
+        try:
+            self.queue.put_nowait(item)
         except Full:
             # silently drop
             pass


Test patch:
diff --git a/tests/test_queues.py b/tests/test_queues.py
--- a/tests/test_queues.py
+++ b/tests/test_queues.py
@@ -89,9 +89,24 @@ def test_multi_processing_handler():
         assert test_handler.has_warning('Hello World')
 
 
+class BatchTestHandler(logbook.TestHandler):
+    def __init__(self, *args, **kwargs):
+        super(BatchTestHandler, self).__init__(*args, **kwargs)
+        self.batches = []
+
+    def emit(self, record):
+        super(BatchTestHandler, self).emit(record)
+        self.batches.append([record])
+
+    def emit_batch(self, records, reason):
+        for record in records:
+            super(BatchTestHandler, self).emit(record)
+        self.batches.append(records)
+
+
 def test_threaded_wrapper_handler(logger):
     from logbook.queues import ThreadedWrapperHandler
-    test_handler = logbook.TestHandler()
+    test_handler = BatchTestHandler()
     with ThreadedWrapperHandler(test_handler) as handler:
         logger.warn('Just testing')
         logger.error('More testing')
@@ -100,6 +115,50 @@ def test_threaded_wrapper_handler(logger):
     handler.close()
 
     assert (not handler.controller.running)
+    assert len(test_handler.records) == 2
+    assert len(test_handler.batches) == 2
+    assert all((len(records) == 1 for records in test_handler.batches))
+    assert test_handler.has_warning('Just testing')
+    assert test_handler.has_error('More testing')
+
+
+def test_threaded_wrapper_handler_emit():
+    from logbook.queues import ThreadedWrapperHandler
+    test_handler = BatchTestHandler()
+    with ThreadedWrapperHandler(test_handler) as handler:
+        lr = logbook.LogRecord('Test Logger', logbook.WARNING, 'Just testing')
+        test_handler.emit(lr)
+        lr = logbook.LogRecord('Test Logger', logbook.ERROR, 'More testing')
+        test_handler.emit(lr)
+
+    # give it some time to sync up
+    handler.close()
+
+    assert (not handler.controller.running)
+    assert len(test_handler.records) == 2
+    assert len(test_handler.batches) == 2
+    assert all((len(records) == 1 for records in test_handler.batches))
+    assert test_handler.has_warning('Just testing')
+    assert test_handler.has_error('More testing')
+
+
+def test_threaded_wrapper_handler_emit_batched():
+    from logbook.queues import ThreadedWrapperHandler
+    test_handler = BatchTestHandler()
+    with ThreadedWrapperHandler(test_handler) as handler:
+        test_handler.emit_batch([
+            logbook.LogRecord('Test Logger', logbook.WARNING, 'Just testing'),
+            logbook.LogRecord('Test Logger', logbook.ERROR, 'More testing'),
+        ], 'group')
+
+    # give it some time to sync up
+    handler.close()
+
+    assert (not handler.controller.running)
+    assert len(test_handler.records) == 2
+    assert len(test_handler.batches) == 1
+    (records, ) = test_handler.batches
+    assert len(records) == 2
     assert test_handler.has_warning('Just testing')
     assert test_handler.has_error('More testing')