Implement WriteLogger (#403)

* Implement WriteLogger

* Fix type hint

* Rewrite tests to use fixtures

* Refactor fixture

* Add to changelog

* PR tweaks

* More tweaks

* [pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

* Unused imports

Co-authored-by: pre-commit-ci[bot] <66853113+pre-commit-ci[bot]@users.noreply.github.com>
Co-authored-by: Hynek Schlawack <hs@ox.cx>

CHANGELOG.rst

@@ -47,6 +47,8 @@ Changes:
   `#396 <https://github.com/hynek/structlog/issues/396>`_
 - ``structlog.PrintLogger`` -- that is used by default -- now uses ``print()`` for printing, making it a better citizen for interactive terminal applications.
   `#399 <https://github.com/hynek/structlog/pull/399>`_
+- Added ``structlog.WriteLogger``, a faster but more low-level alternative to ``structlog.PrintLogger``.
+  `#403 <https://github.com/hynek/structlog/pull/403>`_
 
 
 ----

src/structlog/_loggers.py

@@ -7,18 +7,20 @@
 Logger wrapper and helper class.
 """
 
+from __future__ import annotations
+
 import copy
 import sys
 import threading
 
 from pickle import PicklingError
 from sys import stderr, stdout
-from typing import IO, Any, BinaryIO, Dict, Optional, TextIO
+from typing import IO, Any, BinaryIO, TextIO
 
 from structlog._utils import until_not_interrupted
 
 
-WRITE_LOCKS: Dict[IO[Any], threading.Lock] = {}
+WRITE_LOCKS: dict[IO[Any], threading.Lock] = {}
 
 
 def _get_lock_for_file(file: IO[Any]) -> threading.Lock:
@@ -53,7 +55,7 @@ class PrintLogger:
        monkeypatchability.
     """
 
-    def __init__(self, file: Optional[TextIO] = None):
+    def __init__(self, file: TextIO | None = None):
         self._file = file or stdout
 
         self._lock = _get_lock_for_file(self._file)
@@ -83,7 +85,7 @@ def __setstate__(self, state: Any) -> None:
 
         self._lock = _get_lock_for_file(self._file)
 
-    def __deepcopy__(self, memodict: Dict[Any, Any] = {}) -> "PrintLogger":
+    def __deepcopy__(self, memodict: dict[Any, Any] = {}) -> PrintLogger:
         """
         Create a new PrintLogger with the same attributes. Similar to pickling.
         """
@@ -127,13 +129,119 @@ class PrintLoggerFactory:
     .. versionadded:: 0.4.0
     """
 
-    def __init__(self, file: Optional[TextIO] = None):
+    def __init__(self, file: TextIO | None = None):
         self._file = file
 
     def __call__(self, *args: Any) -> PrintLogger:
         return PrintLogger(self._file)
 
 
+class WriteLogger:
+    """
+    Write events into a file.
+
+    :param file: File to print to. (default: `sys.stdout`)
+
+    >>> from structlog import WriteLogger
+    >>> WriteLogger().msg("hello")
+    hello
+
+    Useful if you follow
+    `current logging best practices <logging-best-practices>`.
+
+    Also very useful for testing and examples since logging is finicky in
+    doctests.
+
+    A little faster and a little less versatile than the PrintLogger.
+
+    .. versionadded:: 22.1
+    """
+
+    def __init__(self, file: TextIO | None = None):
+        self._file = file or sys.stdout
+        self._write = self._file.write
+        self._flush = self._file.flush
+
+        self._lock = _get_lock_for_file(self._file)
+
+    def __getstate__(self) -> str:
+        """
+        Our __getattr__ magic makes this necessary.
+        """
+        if self._file is stdout:
+            return "stdout"
+
+        elif self._file is stderr:
+            return "stderr"
+
+        raise PicklingError(
+            "Only WriteLoggers to sys.stdout and sys.stderr can be pickled."
+        )
+
+    def __setstate__(self, state: Any) -> None:
+        """
+        Our __getattr__ magic makes this necessary.
+        """
+        if state == "stdout":
+            self._file = stdout
+        else:
+            self._file = stderr
+
+        self._lock = _get_lock_for_file(self._file)
+
+    def __deepcopy__(self, memodict: dict[Any, Any] = {}) -> WriteLogger:
+        """
+        Create a new WriteLogger with the same attributes. Similar to pickling.
+        """
+        if self._file not in (sys.stdout, sys.stderr):
+            raise copy.error(
+                "Only WriteLoggers to sys.stdout and sys.stderr "
+                "can be deepcopied."
+            )
+
+        newself = self.__class__(self._file)
+
+        newself._write = newself._file.write
+        newself._flush = newself._file.flush
+        newself._lock = _get_lock_for_file(newself._file)
+
+        return newself
+
+    def __repr__(self) -> str:
+        return f"<WriteLogger(file={self._file!r})>"
+
+    def msg(self, message: str) -> None:
+        """
+        Write and flush *message*.
+        """
+        with self._lock:
+            until_not_interrupted(self._write, message + "\n")
+            until_not_interrupted(self._flush)
+
+    log = debug = info = warn = warning = msg
+    fatal = failure = err = error = critical = exception = msg
+
+
+class WriteLoggerFactory:
+    r"""
+    Produce `WriteLogger`\ s.
+
+    To be used with `structlog.configure`\ 's ``logger_factory``.
+
+    :param file: File to print to. (default: `sys.stdout`)
+
+    Positional arguments are silently ignored.
+
+    .. versionadded:: 22.1
+    """
+
+    def __init__(self, file: TextIO | None = None):
+        self._file = file
+
+    def __call__(self, *args: Any) -> WriteLogger:
+        return WriteLogger(self._file)
+
+
 class BytesLogger:
     r"""
     Writes bytes into a file.
@@ -149,7 +257,7 @@ class BytesLogger:
     """
     __slots__ = ("_file", "_write", "_flush", "_lock")
 
-    def __init__(self, file: Optional[BinaryIO] = None):
+    def __init__(self, file: BinaryIO | None = None):
         self._file = file or sys.stdout.buffer
         self._write = self._file.write
         self._flush = self._file.flush
@@ -183,7 +291,7 @@ def __setstate__(self, state: Any) -> None:
         self._flush = self._file.flush
         self._lock = _get_lock_for_file(self._file)
 
-    def __deepcopy__(self, memodict: Dict[Any, Any] = {}) -> "BytesLogger":
+    def __deepcopy__(self, memodict: dict[Any, Any] = {}) -> BytesLogger:
         """
         Create a new BytesLogger with the same attributes. Similar to pickling.
         """
@@ -230,7 +338,7 @@ class BytesLoggerFactory:
     """
     __slots__ = ("_file",)
 
-    def __init__(self, file: Optional[BinaryIO] = None):
+    def __init__(self, file: BinaryIO | None = None):
         self._file = file
 
     def __call__(self, *args: Any) -> BytesLogger:

tests/test_loggers.py

@@ -17,114 +17,125 @@
     BytesLoggerFactory,
     PrintLogger,
     PrintLoggerFactory,
+    WriteLogger,
+    WriteLoggerFactory,
 )
 
 from .utils import stdlib_log_methods
 
 
-class TestPrintLogger:
-    def test_prints_to_stdout_by_default(self, capsys):
+class TestLoggers:
+    """
+    Tests common to the Print and WriteLoggers.
+    """
+
+    @pytest.fixture(name="logger_cls", params=(WriteLogger, PrintLogger))
+    @staticmethod
+    def _logger_cls(request):
+        return request.param
+
+    def test_prints_to_stdout_by_default(self, logger_cls, capsys):
         """
         Instantiating without arguments gives conveniently a logger to standard
         out.
         """
-        PrintLogger().msg("hello")
+        logger_cls().msg("hello")
 
         out, err = capsys.readouterr()
         assert "hello\n" == out
         assert "" == err
 
-    def test_prints_to_correct_file(self, tmpdir, capsys):
+    def test_prints_to_correct_file(self, logger_cls, tmpdir, capsys):
         """
         Supplied files are respected.
         """
         f = tmpdir.join("test.log")
         fo = f.open("w")
-        PrintLogger(fo).msg("hello")
+        logger_cls(fo).msg("hello")
         out, err = capsys.readouterr()
 
         assert "" == out == err
         fo.close()
         assert "hello\n" == f.read()
 
-    def test_repr(self):
-        """
-        __repr__ makes sense.
-        """
-        assert repr(PrintLogger()).startswith("<PrintLogger(file=")
-
-    def test_lock(self, sio):
+    def test_lock(self, logger_cls, sio):
         """
         Creating a logger adds a lock to WRITE_LOCKS.
         """
         assert sio not in WRITE_LOCKS
 
-        PrintLogger(sio)
+        logger_cls(sio)
 
         assert sio in WRITE_LOCKS
 
     @pytest.mark.parametrize("method", stdlib_log_methods)
-    def test_stdlib_methods_support(self, method, sio):
+    def test_stdlib_methods_support(self, logger_cls, method, sio):
         """
-        PrintLogger implements methods of stdlib loggers.
+        Print/WriteLogger implements methods of stdlib loggers.
         """
-        getattr(PrintLogger(sio), method)("hello")
+        getattr(logger_cls(sio), method)("hello")
 
         assert "hello" in sio.getvalue()
 
     @pytest.mark.parametrize("file", [None, sys.stdout, sys.stderr])
     @pytest.mark.parametrize("proto", range(pickle.HIGHEST_PROTOCOL + 1))
-    def test_pickle(self, file, proto):
+    def test_pickle(self, logger_cls, file, proto):
         """
         Can be pickled and unpickled for stdout and stderr.
 
         Can't compare output because capsys et all would confuse the logic.
         """
-        pl = PrintLogger(file=file)
+        pl = logger_cls(file=file)
 
         rv = pickle.loads(pickle.dumps(pl, proto))
 
         assert pl._file is rv._file
         assert pl._lock is rv._lock
 
     @pytest.mark.parametrize("proto", range(pickle.HIGHEST_PROTOCOL + 1))
-    def test_pickle_not_stdout_stderr(self, tmpdir, proto):
+    def test_pickle_not_stdout_stderr(self, logger_cls, tmpdir, proto):
         """
-        PrintLoggers with different files than stdout/stderr raise a
+        Loggers with different files than stdout/stderr raise a
         PickingError.
         """
         f = tmpdir.join("file.log")
         f.write("")
-        pl = PrintLogger(file=f.open())
+        pl = logger_cls(file=f.open())
 
-        with pytest.raises(pickle.PicklingError, match="Only PrintLoggers to"):
+        with pytest.raises(pickle.PicklingError, match="Only (.+)Loggers to"):
             pickle.dumps(pl, proto)
 
-    def test_deepcopy(self, capsys):
+    def test_deepcopy(self, logger_cls, capsys):
         """
-        Deepcopied PrintLogger works.
+        Deepcopied logger works.
         """
-        copied_logger = copy.deepcopy(PrintLogger())
+        copied_logger = copy.deepcopy(logger_cls())
         copied_logger.msg("hello")
 
         out, err = capsys.readouterr()
         assert "hello\n" == out
         assert "" == err
 
-    def test_deepcopy_no_stdout(self, tmp_path):
+    def test_deepcopy_no_stdout(self, logger_cls, tmp_path):
         """
-        Only PrintLoggers that log to stdout or stderr can be deepcopy-ed.
+        Only loggers that log to stdout or stderr can be deepcopy-ed.
         """
         p = tmp_path / "log.txt"
         with p.open(mode="w") as f:
-            logger = PrintLogger(f)
+            logger = logger_cls(f)
             logger.msg("hello")
 
             with pytest.raises(copy.error):
                 copy.deepcopy(logger)
 
         assert "hello\n" == p.read_text()
 
+    def test_repr(self, logger_cls):
+        """
+        __repr__ makes sense.
+        """
+        assert repr(logger_cls()).startswith(f"<{logger_cls.__name__}(file=")
+
     def test_stdout_monkeypatch(self, monkeypatch, capsys):
         """
         If stdout gets monkeypatched, the new instance receives the output.
@@ -165,6 +176,31 @@ def test_ignores_args(self):
         PrintLoggerFactory()(1, 2, 3)
 
 
+class TestWriteLoggerFactory:
+    def test_does_not_cache(self):
+        """
+        Due to doctest weirdness, we must not re-use WriteLoggers.
+        """
+        f = WriteLoggerFactory()
+
+        assert f() is not f()
+
+    def test_passes_file(self):
+        """
+        If a file is passed to the factory, it get passed on to the logger.
+        """
+        pl = WriteLoggerFactory(sys.stderr)()
+
+        assert sys.stderr is pl._file
+
+    def test_ignores_args(self):
+        """
+        WriteLogger doesn't take positional arguments.  If any are passed to
+        the factory, they are not passed to the logger.
+        """
+        WriteLoggerFactory()(1, 2, 3)
+
+
 class TestBytesLogger:
     def test_prints_to_stdout_by_default(self, capsys):
         """