Coverage for nexios\logging.py: 67%

1from __future__ import annotations 

2import logging 

3import sys 

4from logging import ( 

5 DEBUG, 

6 ERROR, 

7 Formatter, 

8 Handler, 

9 INFO, 

10 Logger, 

11 LogRecord, 

12 NOTSET, 

13 StreamHandler, 

14 getLogger, 

15) 

16from logging.handlers import QueueHandler, QueueListener, RotatingFileHandler 

17from queue import SimpleQueue as Queue 

18from typing import TYPE_CHECKING, Optional, Tuple 

19 

20if TYPE_CHECKING: 

21 from .application import NexiosApp 

22 

23 

24class LocalQueueHandler(QueueHandler): 

25 """Custom QueueHandler to avoid unnecessary record preparation. 

26 

27 Since we use an in-process queue, there's no need to prepare records, 

28 reducing logging overhead. 

29 """ 

30 

31 def prepare(self, record: LogRecord) -> LogRecord: 

32 return record 

33 

34 

35def _setup_logging_queue(*handlers: Handler) -> QueueHandler: 

36 """Creates a LocalQueueHandler and starts a QueueListener.""" 

37 queue: Queue[LogRecord] = Queue() 

38 queue_handler = LocalQueueHandler(queue) 

39 

40 listener = QueueListener(queue, *handlers, respect_handler_level=True) 

41 listener.start() 

42 

43 return queue_handler 

44 

45 

46def has_level_handler(logger: Logger) -> bool: 

47 """Checks if the logger already has an appropriate handler.""" 

48 level = logger.getEffectiveLevel() 

49 current_logger: Optional[Logger] = logger 

50 

51 while current_logger: 

52 if any(handler.level <= level for handler in current_logger.handlers): 

53 return True 

54 if not current_logger.propagate: 

55 break 

56 current_logger = current_logger.parent 

57 

58 return False 

59 

60 

61def create_logger( 

62 logger_name: str = "nexios", 

63 log_level: int = DEBUG, 

64 log_file: Optional[str] = None, 

65 max_bytes: int = 10 * 1024 * 1024, # 10MB per log file 

66 backup_count: int = 5, 

67) -> Logger: 

68 """Creates a high-performance, configurable logger for Nexios. 

69 

70 Args: 

71 logger_name (str): The name of the logger. 

72 log_level (int): The logging level (DEBUG, INFO, ERROR, etc.). 

73 log_file (Optional[str]): Path to a log file for file-based logging. 

74 max_bytes (int): Max size of each log file before rotating. 

75 backup_count (int): Number of backup log files to keep. 

76 

77 Returns: 

78 Logger: A configured logger instance. 

79 """ 

80 logger = getLogger(logger_name) 

81 logger.setLevel(log_level) 

82 

83 console_handler = StreamHandler(sys.stderr) 

84 console_handler.setFormatter( 

85 Formatter("[%(asctime)s] %(levelname)s in %(module)s: %(message)s") 

86 ) 

87 

88 handlers: Tuple[Handler, ...] = (console_handler,) 

89 

90 if log_file: 

91 file_handler = RotatingFileHandler( 

92 log_file, maxBytes=max_bytes, backupCount=backup_count 

93 ) 

94 file_handler.setFormatter( 

95 Formatter("[%(asctime)s] %(levelname)s in %(module)s: %(message)s") 

96 ) 

97 handlers += (file_handler,) 

98 

99 if not has_level_handler(logger): 

100 queue_handler = _setup_logging_queue(*handlers) 

101 logger.addHandler(queue_handler) 

102 

103 return logger