Coverage for pycommons / io / console.py: 100%

1"""The `logger` routine for writing a log string to stdout.""" 

2import datetime 

3from contextlib import AbstractContextManager, nullcontext 

4from typing import Callable, Final 

5 

6from pycommons.processes.caller import is_doc_test 

7 

8#: the "now" function 

9__DTN: Final[Callable[[], datetime.datetime]] = datetime.datetime.now 

10 

11 

12def logger(message: str, note: str = "", 

13 lock: AbstractContextManager = nullcontext(), 

14 do_print: bool = not is_doc_test()) -> None: 

15 """ 

16 Write a message to the console log. 

17 

18 The line starts with the current date and time, includes the note, and 

19 then the message string after an ": ". 

20 This function can use a `lock` context to prevent multiple processes or 

21 threads to write to the console at the same time. 

22 

23 :param message: the message 

24 :param note: a note to put between the time and the message 

25 :param lock: the lock to prevent multiple threads to write log 

26 output at the same time 

27 :param do_print: really print the output, by default `False` if this 

28 method is called from a "doctest", `True` otherwise 

29 

30 >>> from io import StringIO 

31 >>> from contextlib import redirect_stdout 

32 >>> sio = StringIO() 

33 >>> dt1 = datetime.datetime.now() 

34 >>> with redirect_stdout(sio): 

35 ... logger("hello world!", do_print=True) 

36 >>> line = sio.getvalue().strip() 

37 >>> print(line[line.index(" ", line.index(" ") + 1) + 1:]) 

38 hello world! 

39 >>> dt2 = datetime.datetime.now() 

40 >>> dtx = datetime.datetime.strptime(line[:26], "%Y-%m-%d %H:%M:%S.%f") 

41 >>> dt1 <= dtx <= dt2 

42 True 

43 

44 >>> sio = StringIO() 

45 >>> with redirect_stdout(sio): 

46 ... logger("hello world!", "note", do_print=True) 

47 >>> line = sio.getvalue().strip() 

48 >>> print(line[line.index("n"):]) 

49 note: hello world! 

50 

51 >>> from contextlib import AbstractContextManager 

52 >>> class T: 

53 ... def __enter__(self): 

54 ... print("x") 

55 ... def __exit__(self, exc_type, exc_val, exc_tb): 

56 ... print("y") 

57 

58 >>> sio = StringIO() 

59 >>> with redirect_stdout(sio): 

60 ... logger("hello world!", "", T(), do_print=True) 

61 >>> sio.seek(0) 

62 0 

63 >>> lines = sio.readlines() 

64 >>> print(lines[0].rstrip()) 

65 x 

66 >>> l = lines[1] 

67 >>> print(l[l.index(" ", l.index(" ") + 1) + 1:].rstrip()) 

68 hello world! 

69 >>> print(lines[2].rstrip()) 

70 y 

71 

72 >>> sio = StringIO() 

73 >>> with redirect_stdout(sio): 

74 ... logger("hello world!", "note", T(), do_print=True) 

75 >>> sio.seek(0) 

76 0 

77 >>> lines = sio.readlines() 

78 >>> print(lines[0].rstrip()) 

79 x 

80 >>> l = lines[1] 

81 >>> print(l[l.index("n"):].rstrip()) 

82 note: hello world! 

83 >>> print(lines[2].rstrip()) 

84 y 

85 

86 >>> logger("hello world") # not printed in doctests 

87 >>> logger("hello world", do_print=False) # not printed anyway 

88 """ 

89 if do_print: 

90 text: Final[str] = f"{__DTN()}{note}: {message}" 

91 with lock: 

92 print(text, flush=True) # noqa