2 # -*- coding: utf-8 -*-
4 # © Copyright 2021-2023, Scott Gasch
7 This is a module that offers an opinionated take on how whole program
8 logging should be initialized and controlled. It uses the standard
9 Python :mod:`logging` but gives you control, via commandline config,
12 * Set the logging default level (debug, info, warning, error, critical)
13 of the whole program (see: :code:`--logging_level`)... and to override
14 the logging level for individual modules/functions based on their names
15 (see :code:`--lmodule`),
16 * define the logging message format (see :code:`--logging_format` and
17 :code:`--logging_date_format`) including easily adding a PID/TID
18 marker on all messages to help with multithreaded debugging
19 (:code:`--logging_debug_threads`) and force module names of code
20 that emits log messages to be included in the format
21 (:code:`--logging_debug_modules`),
22 * control the destination of logged messages:
24 - log to the console/stderr (:code:`--logging_console`) and/or
25 - log to a rotated file (:code:`--logging_filename`,
26 :code:`--logging_filename_maxsize` and :code:`--logging_filename_count`)
28 - log to the UNIX syslog (:code:`--logging_syslog` and
29 :code:`--logging_syslog_facility`)
31 * optionally squelch repeated messages (:code:`--logging_squelch_repeats`),
32 * optionally log probalistically (:code:`--logging_probabilistically`),
33 * capture printed messages into the info log (:code:`--logging_captures_prints`),
34 * and optionally clear unwanted logging handlers added by other imports
35 before this one (:code:`--logging_clear_preexisting_handlers`).
37 To use this functionality, call :meth:`initialize_logging` early
38 in your program entry point. If you use the
39 :meth:`pyutils.bootstrap.initialize` decorator on your program's entry
40 point, it will call this for you automatically.
52 from logging.config import fileConfig
53 from logging.handlers import RotatingFileHandler, SysLogHandler
54 from typing import Any, Callable, Dict, Iterable, List, Optional
57 from overrides import overrides
59 # This module is commonly used by others in here and should avoid
60 # taking any unnecessary dependencies back on them.
61 from pyutils import argparse_utils, config, misc_utils
63 cfg = config.add_commandline_args(f"Logging ({__file__})", "Args related to logging")
65 "--logging_config_file",
66 type=argparse_utils.valid_filename,
69 help="Config file containing the logging setup, see: https://docs.python.org/3/howto/logging.html#logging-advanced-tutorial",
75 choices=["NOTSET", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
77 help="The global default level below which to squelch log messages; see also --lmodule",
83 help="The format for lines logged via the logger module. See: https://docs.python.org/3/library/logging.html#formatter-objects",
86 "--logging_date_format",
88 default="%Y/%m/%dT%H:%M:%S.%f%z",
90 help="The format of any dates in --logging_format.",
94 action=argparse_utils.ActionNoYes,
96 help="Should we log to the console (stderr)",
103 help="The filename of the logfile to write.",
106 "--logging_filename_maxsize",
108 default=(1024 * 1024),
110 help="The maximum size (in bytes) to write to the logging_filename.",
113 "--logging_filename_count",
117 help="The number of logging_filename copies to keep before deleting.",
121 action=argparse_utils.ActionNoYes,
123 help="Should we log to localhost's syslog.",
126 "--logging_syslog_facility",
152 metavar="SYSLOG_FACILITY_LIST",
153 help="The default syslog message facility identifier",
156 "--logging_debug_threads",
157 action=argparse_utils.ActionNoYes,
159 help="Should we prepend pid/tid data to all log messages?",
162 "--logging_debug_modules",
163 action=argparse_utils.ActionNoYes,
165 help="Should we prepend module/function data to all log messages?",
168 "--logging_info_is_print",
169 action=argparse_utils.ActionNoYes,
171 help="logging.info also prints to stdout.",
174 "--logging_squelch_repeats",
175 action=argparse_utils.ActionNoYes,
177 help="Do we allow code to indicate that it wants to squelch repeated logging messages or should we always log?",
180 "--logging_probabilistically",
181 action=argparse_utils.ActionNoYes,
183 help="Do we allow probabilistic logging (for code that wants it) or should we always log?",
185 # See also: OutputMultiplexer
187 "--logging_captures_prints",
188 action=argparse_utils.ActionNoYes,
190 help="When calling print, also log.info automatically.",
195 metavar="<SCOPE>=<LEVEL>[,<SCOPE>=<LEVEL>...]",
197 "Allows per-scope logging levels which override the global level set with --logging-level."
198 + "Pass a space separated list of <scope>=<level> where <scope> is one of: module, "
199 + "module:function, or :function and <level> is a logging level (e.g. INFO, DEBUG...)"
203 "--logging_clear_preexisting_handlers",
204 action=argparse_utils.ActionNoYes,
207 "Should logging code clear preexisting global logging handlers and thus insist that is "
208 + "alone can add handlers. Use this to work around annoying modules that insert global "
209 + "handlers with formats and logging levels you might now want. Caveat emptor, this may "
210 + "cause you to miss logging messages."
214 BUILT_IN_PRINT = print
215 LOGGING_INITIALIZED = False
218 # A map from logging_callsite_id -> count of logged messages.
219 squelched_logging_counts: Dict[str, int] = {}
222 def squelch_repeated_log_messages(squelch_after_n_repeats: int) -> Callable:
224 A decorator that marks a function as interested in having the logging
225 messages that it produces be squelched (ignored) after it logs the
226 same message more than N times.
230 This decorator affects *ALL* logging messages produced
231 within the decorated function. That said, messages must be
232 identical in order to be squelched. For example, if the same line
233 of code produces different messages (because of, e.g., a format
234 string), the messages are considered to be different.
236 An example of this from the pyutils code itself can be found in
237 :meth:`pyutils.ansi.fg` and :meth:`pyutils.ansi.bg` methods::
239 @logging_utils.squelch_repeated_log_messages(1)
241 name: Optional[str] = "",
242 red: Optional[int] = None,
243 green: Optional[int] = None,
244 blue: Optional[int] = None,
246 force_16color: bool = False,
247 force_216color: bool = False,
251 These methods log stuff like "Using 24-bit color strategy" which
252 gets old really fast and fills up the logs. By decorating the methods
253 with :code:`@logging_utils.squelch_repeated_log_messages(1)` the code
254 is requesting that its logged messages be dropped silently after the
255 first one is produced (note the argument 1).
257 Users can insist that all logged messages always be reflected in the
258 logs using the :code:`--no_logging_squelch_repeats` flag but the default
259 behavior is to allow code to request it be squelched.
261 :code:`--logging_squelch_repeats` only affects code with this decorator
262 on it; it ignores all other code.
265 squelch_after_n_repeats: the number of repeated messages allowed to
266 log before subsequent messages are silently dropped.
269 def squelch_logging_wrapper(f: Callable):
270 from pyutils import function_utils
272 identifier = function_utils.function_identifier(f)
273 squelched_logging_counts[identifier] = squelch_after_n_repeats
276 return squelch_logging_wrapper
279 class SquelchRepeatedMessagesFilter(logging.Filter):
280 """A filter that only logs messages from a given site with the same
281 (exact) message at the same logging level N times and ignores
282 subsequent attempts to log.
284 This filter only affects logging messages that repeat more than a
285 threshold number of times from functions that are tagged with the
286 :code:`@logging_utils.squelched_logging_ok` decorator (see above);
287 all others are ignored.
289 This functionality is enabled by default but can be disabled via
290 the :code:`--no_logging_squelch_repeats` commandline flag.
293 def __init__(self) -> None:
295 self.counters: collections.Counter = collections.Counter()
298 def filter(self, record: logging.LogRecord) -> bool:
299 """Should we drop this log message?"""
300 id1 = f"{record.module}:{record.funcName}"
301 if id1 not in squelched_logging_counts:
303 threshold = squelched_logging_counts[id1]
304 logsite = f"{record.pathname}+{record.lineno}+{record.levelno}+{record.msg}"
305 count = self.counters[logsite]
306 self.counters[logsite] += 1
307 return count < threshold
310 class DynamicPerScopeLoggingLevelFilter(logging.Filter):
311 """This filter only allows logging messages from an allow list of
312 module names or `module:function` names. Blocks all others. This
313 filter is used to implement the :code:`--lmodule` commandline option.
317 You probably don't need to use this directly, just use
318 :code:`--lmodule`. For example, to set logging level to INFO
319 everywhere except "module:function" where it should be DEBUG::
321 # myprogram.py --logging_level=INFO --lmodule=module:function=DEBUG
326 def level_name_to_level(name: str) -> int:
327 """Given a level name, return its numberic value."""
328 numeric_level = getattr(logging, name, None)
329 if not isinstance(numeric_level, int):
330 raise ValueError(f"Invalid level: {name}")
335 default_logging_level: int,
336 per_scope_logging_levels: Optional[str],
338 """Construct the Filter.
341 default_logging_level: the logging level of the whole program
342 per_scope_logging_levels: optional, comma separated overrides of
343 logging level per scope of the format scope=level where
344 scope is of the form "module:function" or ":function" and
345 level is one of NOTSET, DEBUG, INFO, WARNING, ERROR or
349 self.valid_levels = set(
350 ["NOTSET", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]
352 self.default_logging_level = default_logging_level
353 self.level_by_scope = {}
354 if per_scope_logging_levels is not None:
355 for chunk in per_scope_logging_levels.split(","):
358 f'Malformed lmodule directive: "{chunk}", missing "=". Ignored.',
363 (scope, level) = chunk.split("=")
366 f'Malformed lmodule directive: "{chunk}". Ignored.',
370 scope = scope.strip()
371 level = level.strip().upper()
372 if level not in self.valid_levels:
374 f'Malformed lmodule directive: "{chunk}", bad level. Ignored.',
380 ] = DynamicPerScopeLoggingLevelFilter.level_name_to_level(level)
383 def filter(self, record: logging.LogRecord) -> bool:
384 """Decides whether or not to log based on an allow list."""
386 # First try to find a logging level by scope (--lmodule)
387 if len(self.level_by_scope) > 0:
391 f"{record.module}:{record.funcName}",
392 f":{record.funcName}",
394 level = self.level_by_scope.get(scope, None)
395 if level is not None:
396 if min_level is None or level < min_level:
399 # If we found one, use it instead of the global default level.
400 if min_level is not None:
401 return record.levelno >= min_level
403 # Otherwise, use the global logging level (--logging_level)
404 return record.levelno >= self.default_logging_level
407 # A map from function_identifier -> probability of logging (0.0%..100.0%)
408 probabilistic_logging_levels: Dict[str, float] = {}
411 def logging_is_probabilistic(probability_of_logging: float) -> Callable:
412 """A decorator that indicates that all logging statements within the
413 scope of a particular (marked via decorator) function are not
414 deterministic (i.e. they do not always unconditionally log) but rather
415 are probabilistic (i.e. they log N% of the time, randomly) when the
416 user passes the :code:`--logging_probabilistically` commandline flag
417 (which is enabled by default).
421 This affects *ALL* logging statements within the marked function.
422 If you want it to only affect a subset of logging statements,
423 log those statements in a separate function that you invoke
424 from within the "too large" scope and mark that separate function
425 with the :code:`logging_is_probabilistic` decorator instead.
427 That this functionality can be disabled (forcing all logged
428 messages to produce output) via the
429 :code:`--no_logging_probabilistically` cmdline argument.
432 def probabilistic_logging_wrapper(f: Callable):
433 from pyutils import function_utils
435 identifier = function_utils.function_identifier(f)
436 probabilistic_logging_levels[identifier] = probability_of_logging
439 return probabilistic_logging_wrapper
442 class ProbabilisticFilter(logging.Filter):
444 A filter that logs messages probabilistically (i.e. randomly at some
445 percent chance). This filter is used with a decorator (see
446 :meth:`logging_is_probabilistic`) to implement the
447 :code:`--logging_probabilistically` commandline flag.
449 This filter only affects logging messages from functions that have
450 been tagged with the `@logging_utils.probabilistic_logging` decorator.
454 def filter(self, record: logging.LogRecord) -> bool:
455 """Should the message be logged?"""
456 identifier = f"{record.module}:{record.funcName}"
457 threshold = probabilistic_logging_levels.get(identifier, 100.0)
458 return misc_utils.execute_probabilistically(threshold)
461 class OnlyInfoFilter(logging.Filter):
462 """A filter that only logs messages produced at the INFO logging
463 level. This is used by the :code:`--logging_info_is_print`
464 commandline option to select a subset of the logging stream to
465 send to a stdout handler.
469 def filter(self, record: logging.LogRecord):
470 return record.levelno == logging.INFO
473 def _get_current_pidtid() -> str:
474 return f"{os.getpid()}/{threading.get_ident()}"
477 LOGGING_PREFIXES_BY_PIDTID = {}
480 def register_thread_logging_prefix(message: Optional[str]) -> None:
481 """A function that allows a thread to register a string that
482 should be prepended to every log message it produces from now on.
483 Relies on the :class:`MillisecondAwareOptionallyPrependingFormatter`
484 being used with the logging Handler which is the default if you're
488 message: the message to prepend to all logging done by the
489 current thread. If None or "", clears any previous
492 pidtid = _get_current_pidtid()
493 if message and len(message) > 0:
494 LOGGING_PREFIXES_BY_PIDTID[pidtid] = message
495 elif pidtid in LOGGING_PREFIXES_BY_PIDTID:
496 del LOGGING_PREFIXES_BY_PIDTID[pidtid]
499 LOGGING_SUFFIXES_BY_PIDTID = {}
502 def register_thread_logging_suffix(message: Optional[str]) -> None:
503 """A function that allows a thread to register a string that
504 should be appended to every log message it produces from now on.
505 Relies on the :class:`MillisecondAwareOptionallyPrependingFormatter`
506 being used with the logging Handler which is the default if you're
510 message: the message to append to all logging done by the
511 current thread. If None or "", clears any previous
514 pidtid = _get_current_pidtid()
515 if message and len(message) > 0:
516 LOGGING_SUFFIXES_BY_PIDTID[pidtid] = message
517 elif pidtid in LOGGING_PREFIXES_BY_PIDTID:
518 del LOGGING_SUFFIXES_BY_PIDTID[pidtid]
521 class MillisecondAwareOptionallyPrependingFormatter(logging.Formatter):
522 """A formatter for adding milliseconds to log messages which, for
523 whatever reason, the default Python logger doesn't do.
525 This formatter also consults a map of pid+tid -> message from this
526 module allowing threads to optionally and automatically prepend a
527 message to all logging data they output. If the pid+tid is not
528 found nothing is prepended. See
529 :meth:`register_thread_logging_prefix` and
530 :meth:`register_thread_logging_suffix` for details.
534 You probably don't need to use this directly but it is
535 wired in under :meth:`initialize_logging` so that the
536 timestamps in log messages have millisecond level
541 converter = datetime.datetime.fromtimestamp # type: ignore
544 def format(self, record):
545 pidtid = _get_current_pidtid()
546 prefix = LOGGING_PREFIXES_BY_PIDTID.get(pidtid, None)
548 record.msg = prefix + record.msg
549 suffix = LOGGING_SUFFIXES_BY_PIDTID.get(pidtid, None)
551 record.msg = record.msg + suffix
552 return super().format(record)
555 def formatTime(self, record, datefmt=None):
556 ct = MillisecondAwareOptionallyPrependingFormatter.converter(
557 record.created, pytz.timezone("US/Pacific")
560 s = ct.strftime(datefmt)
562 t = ct.strftime("%Y-%m-%d %H:%M:%S")
563 s = f"{t},{record.msecs:%03d}"
567 def _log_about_logging(
569 default_logging_level,
570 preexisting_handlers_count,
574 """This is invoked automatically after logging is initialized such
575 that the first messages in the log are about how logging itself
578 level_name = logging._levelToName.get(
579 default_logging_level, str(default_logging_level)
581 logger.debug("Initialized global logging; logging level is %s.", level_name)
583 config.config["logging_clear_preexisting_handlers"]
584 and preexisting_handlers_count > 0
587 "Logging cleared %d global handlers (--logging_clear_preexisting_handlers)",
588 preexisting_handlers_count,
590 logger.debug('Logging format specification is "%s"', fmt)
591 if config.config["logging_debug_threads"]:
593 "...Logging format spec captures tid/pid. (--logging_debug_threads)"
595 if config.config["logging_debug_modules"]:
597 "...Logging format spec captures files/functions/lineno. (--logging_debug_modules)"
599 if config.config["logging_syslog"]:
601 "Logging to syslog as %s with priority mapping based on level. (--logging_syslog)",
604 if config.config["logging_filename"]:
606 'Logging to file "%s". (--logging_filename)',
607 config.config["logging_filename"],
610 "...with %d bytes max file size. (--logging_filename_maxsize)",
611 config.config["logging_filename_maxsize"],
614 "...and %d rotating backup file count. (--logging_filename_count)",
615 config.config["logging_filename_count"],
617 if config.config["logging_console"]:
618 logger.debug("Logging to the console (stderr). (--logging_console)")
619 if config.config["logging_info_is_print"]:
621 "Logging logger.info messages will be repeated on stdout. (--logging_info_is_print)"
623 if config.config["logging_squelch_repeats"]:
625 "Logging code allowed to request repeated messages be squelched. (--logging_squelch_repeats)"
629 "Logging code forbidden to request messages be squelched; all messages logged. (--no_logging_squelch_repeats)"
631 if config.config["logging_probabilistically"]:
633 "Logging code is allowed to request probabilistic logging. (--logging_probabilistically)"
637 "Logging code is forbidden to request probabilistic logging; messages always logged. (--no_logging_probabilistically)"
639 if config.config["lmodule"]:
641 f'Logging dynamic per-module logging enabled. (--lmodule={config.config["lmodule"]})'
643 if config.config["logging_captures_prints"]:
645 "Logging will capture printed data as logger.info messages. (--logging_captures_prints)"
649 def initialize_logging(logger=None) -> logging.Logger:
650 """Initialize logging for the program. See module level comments
651 for information about what functionality this provides and how to
652 enable or disable functionality via the commandline.
655 :meth:`bootstrap.initialize` decorator on your program's entry point,
656 it will call this for you. See :meth:`pyutils.bootstrap.initialize`
659 global LOGGING_INITIALIZED
660 if LOGGING_INITIALIZED:
661 return logging.getLogger()
662 LOGGING_INITIALIZED = True
665 logger = logging.getLogger()
667 # --logging_clear_preexisting_handlers removes logging handlers
668 # that were registered by global statements during imported module
670 preexisting_handlers_count = 0
671 assert config.has_been_parsed()
672 if config.config["logging_clear_preexisting_handlers"]:
673 while logger.hasHandlers():
674 logger.removeHandler(logger.handlers[0])
675 preexisting_handlers_count += 1
677 # --logging_config_file pulls logging settings from a config file
678 # skipping the rest of this setup.
679 if config.config["logging_config_file"] is not None:
680 fileConfig(config.config["logging_config_file"])
683 handlers: List[logging.Handler] = []
684 handler: Optional[logging.Handler] = None
686 # Global default logging level (--logging_level); messages below
687 # this level will be silenced.
688 logging_level = config.config["logging_level"]
690 logging_level = logging_level.upper()
691 default_logging_level = getattr(logging, logging_level, None)
692 if not isinstance(default_logging_level, int):
693 raise ValueError(f'Invalid level: {config.config["logging_level"]}')
695 # Custom or default --logging_format?
696 if config.config["logging_format"]:
697 fmt = config.config["logging_format"]
698 elif config.config["logging_syslog"]:
699 fmt = "%(levelname).1s:%(filename)s[%(process)d]: %(message)s"
701 fmt = "%(levelname).1s:%(asctime)s: %(message)s"
703 # --logging_debug_threads and --logging_debug_modules both affect
704 # the format by prepending information about the pid/tid or
706 if config.config["logging_debug_threads"]:
707 fmt = f"%(process)d.%(thread)d|{fmt}"
708 if config.config["logging_debug_modules"]:
709 fmt = f"%(filename)s:%(funcName)s:%(lineno)s|{fmt}"
711 # --logging_syslog (optionally with --logging_syslog_facility)
712 # sets up for logging to use the standard system syslogd as a
715 if config.config["logging_syslog"]:
716 if sys.platform not in ("win32", "cygwin"):
717 if config.config["logging_syslog_facility"]:
718 facility_name = "LOG_" + config.config["logging_syslog_facility"]
719 facility = SysLogHandler.__dict__.get(facility_name, SysLogHandler.LOG_USER) # type: ignore
720 assert facility is not None
721 handler = SysLogHandler(facility=facility, address="/dev/log")
722 handler.setFormatter(
723 MillisecondAwareOptionallyPrependingFormatter(
725 datefmt=config.config["logging_date_format"],
728 handlers.append(handler)
730 # --logging_filename (with friends --logging_filename_count and
731 # --logging_filename_maxsize) set up logging to a file on the
732 # filesystem with automatic rotation when it gets too big.
733 if config.config["logging_filename"]:
734 max_bytes = config.config["logging_filename_maxsize"]
735 assert max_bytes and isinstance(max_bytes, int)
736 backup_count = config.config["logging_filename_count"]
737 assert backup_count and isinstance(backup_count, int)
738 handler = RotatingFileHandler(
739 config.config["logging_filename"],
741 backupCount=backup_count,
743 handler.setFormatter(
744 MillisecondAwareOptionallyPrependingFormatter(
746 datefmt=config.config["logging_date_format"],
749 handlers.append(handler)
751 # --logging_console is, ahem, logging to the console.
752 if config.config["logging_console"]:
753 handler = logging.StreamHandler(sys.stderr)
754 handler.setFormatter(
755 MillisecondAwareOptionallyPrependingFormatter(
757 datefmt=config.config["logging_date_format"],
760 handlers.append(handler)
762 if len(handlers) == 0:
763 handlers.append(logging.NullHandler())
764 for handler in handlers:
765 logger.addHandler(handler)
767 # --logging_info_is_print echoes any message to logger.info(x) as
768 # a print statement on stdout.
769 if config.config["logging_info_is_print"]:
770 handler = logging.StreamHandler(sys.stdout)
771 handler.addFilter(OnlyInfoFilter())
772 logger.addHandler(handler)
774 # --logging_squelch_repeats allows code to request repeat logging
775 # messages (identical log site and message contents) to be
776 # silenced. Logging code must request this explicitly, it isn't
777 # automatic. This option just allows the silencing to happen.
778 if config.config["logging_squelch_repeats"]:
779 for handler in handlers:
780 handler.addFilter(SquelchRepeatedMessagesFilter())
782 # --logging_probabilistically allows code to request
783 # non-deterministic logging where messages have some probability
784 # of being produced. Logging code must request this explicitly.
785 # This option just allows the non-deterministic behavior to
786 # happen. Disabling it will cause every log message to be
788 if config.config["logging_probabilistically"]:
789 for handler in handlers:
790 handler.addFilter(ProbabilisticFilter())
792 # --lmodule is a way to have a special logging level for just on
793 # module or one set of modules that is different than the one set
794 # globally via --logging_level.
795 for handler in handlers:
797 DynamicPerScopeLoggingLevelFilter(
798 default_logging_level,
799 config.config["lmodule"],
803 logger.propagate = False
805 # --logging_captures_prints, if set, will capture and log.info
806 # anything printed on stdout.
807 if config.config["logging_captures_prints"]:
810 def print_and_also_log(*arg, **kwarg):
811 f = kwarg.get("file", None)
816 BUILT_IN_PRINT(*arg, **kwarg)
818 builtins.print = print_and_also_log
820 # At this point the logger is ready, handlers are set up,
821 # etc... so log about the logging configuration.
824 default_logging_level,
825 preexisting_handlers_count,
832 def get_logger(name: str = ""):
833 """Get the global logger"""
834 logger = logging.getLogger(name)
835 return initialize_logging(logger)
838 def tprint(*args, **kwargs) -> None:
839 """Legacy function for printing a message augmented with thread id
840 still needed by some code. Please use --logging_debug_threads in
843 if config.config["logging_debug_threads"]:
844 from pyutils.parallelize.thread_utils import current_thread_id
846 print(f"{current_thread_id()}", end="")
847 print(*args, **kwargs)
852 class OutputMultiplexer(object):
853 """A class that broadcasts printed messages to several sinks
854 (including various logging levels, different files, different file
855 handles, the house log, etc...). See also
856 :class:`OutputMultiplexerContext` for an easy usage pattern.
859 class Destination(enum.IntEnum):
860 """Bits in the destination_bitv bitvector. Used to indicate the
861 output destination."""
866 LOG_WARNING = 0x04 # ⎬ Must provide logger to the c'tor.
868 LOG_CRITICAL = 0x10 # ⎭
869 FILENAMES = 0x20 # Must provide a filename to the c'tor.
870 FILEHANDLES = 0x40 # Must provide a handle to the c'tor.
872 ALL_LOG_DESTINATIONS = (
873 LOG_DEBUG | LOG_INFO | LOG_WARNING | LOG_ERROR | LOG_CRITICAL
875 ALL_OUTPUT_DESTINATIONS = 0x8F
880 destination_bitv: int,
883 filenames: Optional[Iterable[str]] = None,
884 handles: Optional[Iterable[io.TextIOWrapper]] = None,
887 Constructs the OutputMultiplexer instance.
890 destination_bitv: a bitvector where each bit represents an
891 output destination. Multiple bits may be set.
892 logger: if LOG_* bits are set, you must pass a logger here.
893 filenames: if FILENAMES bit is set, this should be a list of
894 files you'd like to output into. This code handles opening
895 and closing said files.
896 handles: if FILEHANDLES bit is set, this should be a list of
897 already opened filehandles you'd like to output into. The
898 handles will remain open after the scope of the multiplexer.
901 logger = logging.getLogger(None)
904 self.f: Optional[List[Any]] = None
905 if filenames is not None:
906 self.f = [open(filename, "wb", buffering=0) for filename in filenames]
908 if destination_bitv & OutputMultiplexer.Destination.FILENAMES:
909 raise ValueError("Filenames argument is required if bitv & FILENAMES")
912 self.h: Optional[List[Any]] = None
913 if handles is not None:
914 self.h = list(handles)
916 if destination_bitv & OutputMultiplexer.Destination.FILEHANDLES:
917 raise ValueError("Handle argument is required if bitv & FILEHANDLES")
920 self.set_destination_bitv(destination_bitv)
922 def get_destination_bitv(self):
923 """Where are we outputting?"""
924 return self.destination_bitv
926 def set_destination_bitv(self, destination_bitv: int):
927 """Change the output destination_bitv to the one provided."""
928 if destination_bitv & self.Destination.FILENAMES and self.f is None:
929 raise ValueError("Filename argument is required if bitv & FILENAMES")
930 if destination_bitv & self.Destination.FILEHANDLES and self.h is None:
931 raise ValueError("Handle argument is required if bitv & FILEHANDLES")
932 self.destination_bitv = destination_bitv
934 def print(self, *args, **kwargs):
935 """Produce some output to all sinks."""
936 from pyutils.string_utils import _sprintf, strip_escape_sequences
938 end = kwargs.pop("end", None)
940 if not isinstance(end, str):
941 raise TypeError("end must be None or a string")
942 sep = kwargs.pop("sep", None)
944 if not isinstance(sep, str):
945 raise TypeError("sep must be None or a string")
947 raise TypeError("invalid keyword arguments to print()")
948 buf = _sprintf(*args, end="", sep=sep)
955 if self.destination_bitv & self.Destination.FILENAMES and self.f is not None:
957 _.write(buf.encode("utf-8"))
960 if self.destination_bitv & self.Destination.FILEHANDLES and self.h is not None:
965 buf = strip_escape_sequences(buf)
966 if self.logger is not None:
967 if self.destination_bitv & self.Destination.LOG_DEBUG:
968 self.logger.debug(buf)
969 if self.destination_bitv & self.Destination.LOG_INFO:
970 self.logger.info(buf)
971 if self.destination_bitv & self.Destination.LOG_WARNING:
972 self.logger.warning(buf)
973 if self.destination_bitv & self.Destination.LOG_ERROR:
974 self.logger.error(buf)
975 if self.destination_bitv & self.Destination.LOG_CRITICAL:
976 self.logger.critical(buf)
977 if self.destination_bitv & self.Destination.HLOG:
981 """Close all open files."""
982 if self.f is not None:
987 class OutputMultiplexerContext(OutputMultiplexer, contextlib.ContextDecorator):
989 A context that uses an :class:`OutputMultiplexer`. e.g.::
991 with OutputMultiplexerContext(
992 OutputMultiplexer.LOG_INFO |
993 OutputMultiplexer.LOG_DEBUG |
994 OutputMultiplexer.FILENAMES |
995 OutputMultiplexer.FILEHANDLES,
996 filenames = [ '/tmp/foo.log', '/var/log/bar.log' ],
999 mplex.print("This is a log message!")
1004 destination_bitv: OutputMultiplexer.Destination,
1012 destination_bitv: a bitvector that indicates where we should
1013 send output. See :class:`OutputMultiplexer` for options.
1014 logger: optional logger to use for log destination messages.
1015 filenames: optional filenames to write for filename destination
1017 handles: optional open filehandles to write for filehandle
1018 destination messages.
1023 filenames=filenames,
1027 def __enter__(self):
1030 def __exit__(self, etype, value, traceback) -> bool:
1032 if etype is not None:
1037 def hlog(message: str) -> None:
1038 """Write a message to the house log (syslog facility local7 priority
1039 info) by calling `/usr/bin/logger`. This is pretty hacky but used
1040 by a bunch of (my) code. Another way to do this would be to use
1041 :code:`--logging_syslog` and :code:`--logging_syslog_facility` but
1042 I can't actually say that's easier.
1044 TODO: this needs to move.
1046 message = message.replace("'", "'\"'\"'")
1047 os.system(f"/usr/bin/logger -p local7.info -- '{message}'")
1050 if __name__ == "__main__":