Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
15 changes: 14 additions & 1 deletion boilerplates/logging.py
Original file line number Diff line number Diff line change
Expand Up @@ -218,12 +218,24 @@ def unittest_verbosity() -> t.Optional[int]:
return None


class StreamToCall:
class StreamToCall(t.IO[str]):
"""Redirect stream writes to a function call.

Enable using logging instances as a file-like objects.
Given a called_function, convert write(text) calls to called_function(text) calls.
For example: StreamToCall(logging.warning) will redirect all writes to logging.warning().

Example usage:

logger = logging.getLogger(__name__)
with contextlib.redirect_stdout(boilerplates.logging.StreamToCall(logger.info)):
print('this will be logged at INFO level')

Another example:

logger = logging.getLogger(__name__)
stream = boilerplates.logging.StreamToCall(logger.debug)
print('this will be logged at DEBUG level', file=stream)
"""

def __init__(self, called_function: collections.abc.Callable):
Expand All @@ -235,6 +247,7 @@ def write(self, message: str, *args):
while message.endswith('\r') or message.endswith('\n'):
message = message[:-1]
self._function(message, *args)
return len(message)

def flush(self):
"""Flush can be a no-op."""
9 changes: 8 additions & 1 deletion test/test_logging.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
"""Unit tests for logging boilerplate."""

import contextlib
import inspect
import logging
import os
Expand Down Expand Up @@ -103,7 +104,13 @@ def test_unittest_verbosity_not_unittest(self):
verbosity = boilerplates.logging.unittest_verbosity()
self.assertIsNone(verbosity)

def test_stream_to_call(self):
def test_stream_to_call_contextlib(self):
log = logging.getLogger(f'{__name__}.test_stream_to_call')
with contextlib.redirect_stdout(boilerplates.logging.StreamToCall(log.info)):
with self.assertLogs(logger=log, level='INFO'):
print('test output')

def test_stream_to_call_direct(self):
log = logging.getLogger(f'{__name__}.test_stream_to_call')
stream = boilerplates.logging.StreamToCall(log.info)
with self.assertLogs(logger=log, level='INFO'):
Expand Down
Loading