3 """Utilities related to logging."""
7 from logging.handlers import RotatingFileHandler, SysLogHandler
13 import string_utils as su
14 import thread_utils as tu
16 parser = config.add_commandline_args(
17 f'Logging ({__file__})',
18 'Args related to logging')
20 '--logging_config_file',
21 type=argparse_utils.valid_filename,
24 help='Config file containing the logging setup, see: https://docs.python.org/3/howto/logging.html#logging-advanced-tutorial',
30 choices=['NOTSET', 'DEBUG', 'INFO', 'WARNING', 'ERROR', 'CRITICAL'],
32 help='The level below which to squelch log messages.',
37 default='%(levelname)s:%(asctime)s: %(message)s',
38 help='The format for lines logged via the logger module.'
41 '--logging_date_format',
43 default='%Y/%m/%dT%H:%M:%S%z',
45 help='The format of any dates in --logging_format.'
49 action=argparse_utils.ActionNoYes,
51 help='Should we log to the console (stderr)',
58 help='The filename of the logfile to write.'
61 '--logging_filename_maxsize',
65 help='The maximum size (in bytes) to write to the logging_filename.'
68 '--logging_filename_count',
72 help='The number of logging_filename copies to keep before deleting.'
76 action=argparse_utils.ActionNoYes,
78 help='Should we log to localhost\'s syslog.'
81 '--logging_debug_threads',
82 action=argparse_utils.ActionNoYes,
84 help='Should we prepend pid/tid data to all log messages?'
87 '--logging_info_is_print',
88 action=argparse_utils.ActionNoYes,
90 help='logging.info also prints to stdout.'
94 class OnlyInfoFilter(logging.Filter):
95 def filter(self, record):
96 return record.levelno == logging.INFO
99 def initialize_logging(logger=None) -> logging.Logger:
100 assert config.has_been_parsed()
102 logger = logging.getLogger() # Root logger
104 if config.config['logging_config_file'] is not None:
105 logging.config.fileConfig('logging.conf')
109 numeric_level = getattr(
111 config.config['logging_level'].upper(),
114 if not isinstance(numeric_level, int):
115 raise ValueError('Invalid level: %s' % config.config['logging_level'])
117 fmt = config.config['logging_format']
118 if config.config['logging_debug_threads']:
119 fmt = f'%(process)d.%(thread)d|{fmt}'
121 if config.config['logging_syslog']:
122 if sys.platform in ('win32', 'cygwin'):
124 "WARNING: Current platform does not support syslog; IGNORING.",
128 handler = SysLogHandler()
129 # for k, v in encoded_priorities.items():
130 # handler.encodePriority(k, v)
131 handler.setFormatter(
134 datefmt=config.config['logging_date_format'],
137 handler.setLevel(numeric_level)
138 handlers.append(handler)
140 if config.config['logging_filename'] is not None:
141 handler = RotatingFileHandler(
142 config.config['logging_filename'],
143 maxBytes = config.config['logging_filename_maxsize'],
144 backupCount = config.config['logging_filename_count'],
146 handler.setLevel(numeric_level)
147 handler.setFormatter(
150 datefmt=config.config['logging_date_format'],
153 handlers.append(handler)
155 if config.config['logging_console']:
156 handler = logging.StreamHandler(sys.stderr)
157 handler.setLevel(numeric_level)
158 handler.setFormatter(
161 datefmt=config.config['logging_date_format'],
164 handlers.append(handler)
166 if len(handlers) == 0:
167 handlers.append(logging.NullHandler())
169 for handler in handlers:
170 logger.addHandler(handler)
171 if config.config['logging_info_is_print']:
172 handler = logging.StreamHandler(sys.stdout)
173 handler.addFilter(OnlyInfoFilter())
174 logger.addHandler(handler)
175 logger.setLevel(numeric_level)
176 logger.propagate = False
180 def get_logger(name: str = ""):
181 logger = logging.getLogger(name)
182 return initialize_logging(logger)
185 def tprint(*args, **kwargs) -> None:
186 if config.config['logging_debug_threads']:
187 print(f'{tu.current_thread_id()}', end="")
188 print(*args, **kwargs)
193 def dprint(*args, **kwargs) -> None:
194 print(*args, file=sys.stderr, **kwargs)
197 class OutputSink(object):
199 # Bits in the destination_bitv bitvector. Used to indicate the
200 # output destination.
205 LOG_WARNING = 0x10 # > Should provide logger to the c'tor.
207 LOG_CRITICAL = 0x40 # _/
208 FILENAME = 0x80 # Must provide a filename to the c'tor.
211 ALL_LOG_DESTINATIONS = (
212 LOG_DEBUG | LOG_INFO | LOG_WARNING | LOG_ERROR | LOG_CRITICAL
214 ALL_OUTPUT_DESTINATIONS = 0x1FF
217 destination_bitv: int,
222 logger = logging.getLogger(None)
225 if filename is not None:
226 self.f = open(filename, "wb", buffering=0)
228 if self.destination_bitv & OutputSink.FILENAME:
230 "Filename argument is required if bitv & FILENAME"
233 self.set_destination_bitv(destination_bitv)
235 def get_destination_bitv(self):
236 return self.destination_bitv
238 def set_destination_bitv(self, destination_bitv: int):
239 if destination_bitv & self.FILENAME and self.f is None:
241 "Filename argument is required if bitv & FILENAME"
243 self.destination_bitv = destination_bitv
245 def print(self, *args, **kwargs):
246 end = kwargs.pop("end", None)
248 if not isinstance(end, str):
249 raise TypeError("end must be None or a string")
250 sep = kwargs.pop("sep", None)
252 if not isinstance(sep, str):
253 raise TypeError("sep must be None or a string")
255 raise TypeError("invalid keyword arguments to print()")
256 buf = su.sprintf(*args, end="", sep=sep)
261 if self.destination_bitv & self.STDOUT:
262 print(buf, file=sys.stdout, sep=sep, end=end)
263 if self.destination_bitv & self.STDERR:
264 print(buf, file=sys.stderr, sep=sep, end=end)
267 if self.destination_bitv & self.FILENAME and self.f is not None:
268 self.f.write(buf.encode('utf-8'))
270 buf = su.strip_escape_sequences(buf)
271 if self.logger is not None:
272 if self.destination_bitv & self.LOG_DEBUG:
273 self.logger.debug(buf)
274 if self.destination_bitv & self.LOG_INFO:
275 self.logger.info(buf)
276 if self.destination_bitv & self.LOG_WARNING:
277 self.logger.warning(buf)
278 if self.destination_bitv & self.LOG_ERROR:
279 self.logger.error(buf)
280 if self.destination_bitv & self.LOG_CRITICAL:
281 self.logger.critical(buf)
282 if self.destination_bitv & self.HLOG:
286 if self.f is not None:
290 class OutputContext(OutputSink, contextlib.ContextDecorator):
292 destination_bitv: int,
296 super().__init__(destination_bitv, logger=logger, filename=filename)
301 def __exit__(self, etype, value, traceback):
303 if etype is not None:
308 def hlog(message: str) -> None:
309 message = message.replace("'", "'\"'\"'")
310 os.system(f"/usr/bin/logger -p local7.info -- '{message}'")