Implement async bound loggers for stdlib

.pre-commit-config.yaml

@@ -21,7 +21,7 @@ repos:
       exclude: docs/code_examples
 
   - repo: https://github.com/pre-commit/pre-commit-hooks
-    rev: v3.3.0
+    rev: v3.4.0
     hooks:
       - id: trailing-whitespace
       - id: end-of-file-fixer

CHANGELOG.rst

@@ -53,7 +53,9 @@ Changes:
   `#266 <https://github.com/hynek/structlog/issues/266>`_,
 - ``structlog.PrintLogger`` now supports ``copy.deepcopy()``.
   `#268 <https://github.com/hynek/structlog/issues/268>`_
-- Added ``structlog.testing.CapturingLogger`` for more unittesting goodness.
+- Added ``structlog.testing.CapturingLogger`` for more unit testing goodness.
+- Added ``structlog.stdlib.AsyncBoundLogger`` that executes logging calls in a thread executor and therefore doesn't block.
+  `#245 <https://github.com/hynek/structlog/pull/245>`_
 
 
 ----

conftest.py

@@ -3,25 +3,41 @@
 # repository for complete details.
 
 
+import logging
 import sys
 
 from io import StringIO
 
 import pytest
 
 from structlog._log_levels import _NAME_TO_LEVEL
+from structlog.testing import CapturingLogger
 
 
 try:
     import twisted
 except ImportError:
     twisted = None
 
+LOGGER = logging.getLogger()
 
-@pytest.fixture
-def sio():
+
+@pytest.fixture(autouse=True)
+def _ensure_logging_framework_not_altered():
+    """
+    Prevents 'ValueError: I/O operation on closed file.' errors.
+    """
+    before_handlers = list(LOGGER.handlers)
+
+    yield
+
+    LOGGER.handlers = before_handlers
+
+
+@pytest.fixture(name="sio")
+def _sio():
     """
-    A StringIO instance.
+    A new StringIO instance.
     """
     return StringIO()
 
@@ -43,10 +59,15 @@ def __repr__(self):
     name="stdlib_log_method",
     params=[m for m in _NAME_TO_LEVEL if m != "notset"],
 )
-def fixture_stdlib_log_methods(request):
+def _stdlib_log_methods(request):
     return request.param
 
 
+@pytest.fixture(name="cl")
+def _cl():
+    return CapturingLogger()
+
+
 collect_ignore = []
 if sys.version_info[:2] < (3, 7):
     collect_ignore.append("tests/test_contextvars.py")

docs/api.rst

@@ -235,7 +235,9 @@ API Reference
 .. autofunction:: get_logger
 
 .. autoclass:: BoundLogger
-   :members: bind, unbind, new, debug, info, warning, warn, error, critical, exception, log
+   :members: bind, unbind, try_unbind, new, debug, info, warning, warn, error, critical, exception, log
+
+.. autoclass:: AsyncBoundLogger
 
 .. autoclass:: LoggerFactory
    :members: __call__

docs/conf.py

@@ -74,6 +74,7 @@ def find_version(*file_paths):
     ("py:class", "TextIO"),
     ("py:class", "structlog._base.BoundLoggerBase"),
     ("py:class", "structlog.dev._Styles"),
+    ("py:obj", "sync_bl"),
 ]
 
 # If true, '()' will be appended to :func: etc. cross-reference text.

docs/standard-library.rst

@@ -48,6 +48,17 @@ Please note though, that it will neither configure nor verify your configuration
 It will call `structlog.get_logger()` just like if you would've called it -- the only difference are the type hints.
 
 
+``asyncio``
+^^^^^^^^^^^
+
+For ``asyncio`` applications, you may not want your whole application to block while your processor chain is formatting your log entries.
+For that use case ``structlog`` comes with `structlog.stdlib.AsyncBoundLogger` that will do all processing in a thread pool executor.
+
+This means an increased computational cost per log entry but your application will never block because of logging.
+
+To use it, :doc:`configure <configuration>` ``structlog`` to use `AsyncBoundLogger` as ``wrapper_class``.
+
+
 Processors
 ----------
 

setup.py

@@ -47,7 +47,7 @@
         "coverage[toml]",
         "freezegun>=0.2.8",
         "pretend",
-        "pytest-asyncio; python_version>='3.7'",
+        "pytest-asyncio",
         "pytest-randomly",
         "pytest>=6.0",
         "simplejson",

src/structlog/stdlib.py

@@ -8,29 +8,48 @@
 
 See also :doc:`structlog's standard library support <standard-library>`.
 """
-import logging
 
-from typing import Any, Dict, List, Optional, Sequence, Tuple
+import asyncio
+import logging
+import sys
+
+from functools import partial
+from typing import (
+    Any,
+    Callable,
+    Dict,
+    Iterable,
+    List,
+    Optional,
+    Sequence,
+    Tuple,
+)
 
 from ._base import BoundLoggerBase
 from ._config import get_logger as _generic_get_logger
 from ._frames import _find_first_app_frame_and_name, _format_stack
 from ._log_levels import _LEVEL_TO_NAME, _NAME_TO_LEVEL, add_log_level
 from .exceptions import DropEvent
-from .types import EventDict, ExcInfo, Processor, WrappedLogger
+from .types import Context, EventDict, ExcInfo, Processor, WrappedLogger
+
+
+try:
+    from . import contextvars
+except ImportError:
+    contextvars = None  # type: ignore
 
 
 __all__ = [
+    "add_log_level_number",
+    "add_log_level",
+    "add_logger_name",
     "BoundLogger",
+    "filter_by_level",
     "get_logger",
     "LoggerFactory",
     "PositionalArgumentsFormatter",
-    "filter_by_level",
-    "add_log_level_number",
-    "add_log_level",
-    "add_logger_name",
-    "render_to_log_kwargs",
     "ProcessorFormatter",
+    "render_to_log_kwargs",
 ]
 
 
@@ -337,6 +356,159 @@ def get_logger(*args: Any, **initial_values: Any) -> BoundLogger:
     return _generic_get_logger(*args, **initial_values)
 
 
+class AsyncBoundLogger:
+    """
+    Wraps a `BoundLogger` & exposes its logging methods as ``async`` versions.
+
+    Instead of blocking the program, they are run asynchronously in a thread
+    pool executor.
+
+    This means more computational overhead per log call. But it also means that
+    the processor chain (e.g. JSON serialization) and I/O won't block your
+    whole application.
+
+    .. warning: Since the processor pipeline runs in a separate thread,
+        `structlog.contextvars.merge_contextvars` does **nothing** and should
+        be removed from you processor chain.
+
+        Instead it's merged within **this logger** before handing off log
+        processing to the thread.
+
+    Only available for Python 3.7 and later.
+
+    :ivar structlog.stdlib.BoundLogger sync_bl: The wrapped synchronous logger.
+       It is useful to be able to log synchronously occasionally.
+
+    .. versionadded:: 20.2.0
+    """
+
+    __slots__ = ["sync_bl", "_loop"]
+
+    sync_bl: BoundLogger
+
+    _executor = None
+    _bound_logger_factory = BoundLogger
+
+    def __init__(
+        self,
+        logger: logging.Logger,
+        processors: Iterable[Processor],
+        context: Context,
+        *,
+        # Only as an optimization for binding!
+        _sync_bl: Any = None,  # *vroom vroom* over purity.
+        _loop: Any = None,
+    ):
+        if _sync_bl:
+            self.sync_bl = _sync_bl
+            self._loop = _loop
+
+            return
+
+        self.sync_bl = self._bound_logger_factory(
+            logger=logger, processors=processors, context=context
+        )
+        self._loop = asyncio.get_running_loop()
+
+    @property
+    def _context(self) -> Context:
+        return self.sync_bl._context
+
+    def bind(self, **new_values: Any) -> "AsyncBoundLogger":
+        return AsyncBoundLogger(
+            # logger, processors and context are within sync_bl. These
+            # arguments are ignored if _sync_bl is passsed. *vroom vroom* over
+            # purity.
+            logger=None,  # type: ignore
+            processors=(),
+            context={},
+            _sync_bl=self.sync_bl.bind(**new_values),
+            _loop=self._loop,
+        )
+
+    def new(self, **new_values: Any) -> "AsyncBoundLogger":
+        return AsyncBoundLogger(
+            # c.f. comment in bind
+            logger=None,  # type: ignore
+            processors=(),
+            context={},
+            _sync_bl=self.sync_bl.new(**new_values),
+            _loop=self._loop,
+        )
+
+    def unbind(self, *keys: str) -> "AsyncBoundLogger":
+        return AsyncBoundLogger(
+            # c.f. comment in bind
+            logger=None,  # type: ignore
+            processors=(),
+            context={},
+            _sync_bl=self.sync_bl.unbind(*keys),
+            _loop=self._loop,
+        )
+
+    def try_unbind(self, *keys: str) -> "AsyncBoundLogger":
+        return AsyncBoundLogger(
+            # c.f. comment in bind
+            logger=None,  # type: ignore
+            processors=(),
+            context={},
+            _sync_bl=self.sync_bl.try_unbind(*keys),
+            _loop=self._loop,
+        )
+
+    async def _dispatch_to_sync(
+        self,
+        meth: Callable[..., Any],
+        event: str,
+        args: Tuple[Any, ...],
+        kw: Dict[str, Any],
+    ) -> None:
+        """
+        Merge contextvars and log using the sync logger in a thread pool.
+        """
+        ctx = contextvars._get_context().copy()
+        ctx.update(kw)
+
+        await self._loop.run_in_executor(
+            self._executor, partial(meth, event, *args, **ctx)
+        )
+
+    async def debug(self, event: str, *args: Any, **kw: Any) -> None:
+        await self._dispatch_to_sync(self.sync_bl.debug, event, args, kw)
+
+    async def info(self, event: str, *args: Any, **kw: Any) -> None:
+        await self._dispatch_to_sync(self.sync_bl.info, event, args, kw)
+
+    async def warning(self, event: str, *args: Any, **kw: Any) -> None:
+        await self._dispatch_to_sync(self.sync_bl.warning, event, args, kw)
+
+    warn = warning
+
+    async def error(self, event: str, *args: Any, **kw: Any) -> None:
+        await self._dispatch_to_sync(self.sync_bl.error, event, args, kw)
+
+    async def critical(self, event: str, *args: Any, **kw: Any) -> None:
+        await self._dispatch_to_sync(self.sync_bl.critical, event, args, kw)
+
+    fatal = critical
+
+    async def exception(self, event: str, *args: Any, **kw: Any) -> None:
+        # To make `log.exception("foo") work, we have to check if the user
+        # passed an explicit exc_info and if not, supply our own.
+        ei = kw.pop("exc_info", None)
+        if ei is None and kw.get("exception") is None:
+            ei = sys.exc_info()
+
+        kw["exc_info"] = ei
+
+        await self._dispatch_to_sync(self.sync_bl.exception, event, args, kw)
+
+    async def log(self, level: Any, event: str, *args: Any, **kw: Any) -> None:
+        await self._dispatch_to_sync(
+            partial(self.sync_bl.log, level), event, args, kw
+        )
+
+
 class LoggerFactory:
     """
     Build a standard library logger when an *instance* is called.

tests/test_loggers.py

@@ -7,7 +7,7 @@
 import pickle
 import sys
 
-from io import BytesIO, StringIO
+from io import BytesIO
 
 import pytest
 
@@ -53,25 +53,21 @@ def test_repr(self):
         """
         assert repr(PrintLogger()).startswith("<PrintLogger(file=")
 
-    def test_lock(self):
+    def test_lock(self, sio):
         """
         Creating a logger adds a lock to WRITE_LOCKS.
         """
-        sio = StringIO()
-
         assert sio not in WRITE_LOCKS
 
         PrintLogger(sio)
 
         assert sio in WRITE_LOCKS
 
     @pytest.mark.parametrize("method", stdlib_log_methods)
-    def test_stdlib_methods_support(self, method):
+    def test_stdlib_methods_support(self, method, sio):
         """
         PrintLogger implements methods of stdlib loggers.
         """
-        sio = StringIO()
-
         getattr(PrintLogger(sio), method)("hello")
 
         assert "hello" in sio.getvalue()
@@ -186,12 +182,10 @@ def test_repr(self):
         """
         assert repr(BytesLogger()).startswith("<BytesLogger(file=")
 
-    def test_lock(self):
+    def test_lock(self, sio):
         """
         Creating a logger adds a lock to WRITE_LOCKS.
         """
-        sio = StringIO()
-
         assert sio not in WRITE_LOCKS
 
         BytesLogger(sio)