Coverage for jaypore_ci/logging.py: 89%

1""" 

2The basic logging module. 

3""" 

4import logging 

5from typing import Any 

6 

7import structlog 

8 

9# This is used to accumulate logs and is later sent over to the CI status as a 

10# separate log list 

11jaypore_logs = [] 

12 

13 

14class JayporeLogger: 

15 """ 

16 This is mainly used to collect logs into a single global variable so that 

17 the logs of the CI runner itself can also be posted as part of the CI 

18 report. 

19 """ 

20 

21 def __getstate__(self) -> str: 

22 return "stdout" 

23 

24 def __setstate__(self, state: Any) -> None: 

25 pass 

26 

27 def __deepcopy__(self, memodict: dict[Any, Any] = None) -> "JayporeLogger": 

28 return self.__class__() 

29 

30 def msg(self, message: str) -> None: 

31 global jaypore_logs # pylint: disable=global-statement 

32 jaypore_logs.append(message) 

33 if len(jaypore_logs) > 1500: 

34 jaypore_logs = jaypore_logs[-1000:] 

35 print(message) 

36 

37 log = debug = info = warn = warning = msg 

38 fatal = failure = err = error = critical = exception = msg 

39 

40 

41class JayporeLoggerFactory: 

42 def __init__(self): 

43 pass 

44 

45 def __call__(self, *args) -> JayporeLogger: 

46 return JayporeLogger() 

47 

48 

49structlog.configure( 

50 processors=[ 

51 structlog.contextvars.merge_contextvars, 

52 structlog.processors.add_log_level, 

53 # structlog.processors.StackInfoRenderer(), 

54 # structlog.dev.set_exc_info, 

55 structlog.processors.TimeStamper(fmt="iso"), 

56 structlog.dev.ConsoleRenderer(colors=False), 

57 ], 

58 wrapper_class=structlog.make_filtering_bound_logger(logging.NOTSET), 

59 context_class=dict, 

60 logger_factory=JayporeLoggerFactory(), 

61 cache_logger_on_first_use=False, 

62) 

63logger = structlog.get_logger()