##########################################################################################
# pdslogger/__init__.py
##########################################################################################
"""PDS RMS Node enhancements to the Python logger module.
The :class:`PdsLogger` class provides a variety of enhancements to Python's
``logging.Logger`` class. Use the :meth:`PdsLogger` constructor or method
:meth:`~PdsLogger.get_logger` to create a new logger or access an existing one by name. To
add the capabilities of :class:`PdsLogger` to an existing ``logging.Logger``, use
:meth:`~PdsLogger.as_pdslogger`. In this case, the behavior of the new :class:`PdsLogger`
will be identical to that of the given ``Logger``, including all formatting, but the
additional capabilities of :class:`PdsLogger` will also become available.
It supports hierarchical logs, in which the log generated by a running task can be cleanly
separated into individual sub-tasks. When a sub-task is completed, the log contains a
summary of the numbers of logged messages by category and, optionally, the elapsed time.
You create a new level in the hierarchy with :meth:`~PdsLogger.open` and close them with
:meth:`~PdsLogger.close`. Alternatively, you can write::
with logger.open(...):
# Write log messages here
to close the newly-opened section of the log automatically. A specific handler can be
assigned to the :class:`PdsLogger` as part of the :meth:`~PdsLogger.open` call and it will
automatically be removed upon closing. Use this feature, for example, to obtain separate
log files for individual tasks or when processing a sequence of individual files.
The constructor allows the user to create additional error categories beyond the standard
ones named "`debug`", "`info`", "`warning`", etc. Each category can be assigned its own
numeric level, where DEBUG=10, INFO=20, WARNING=30, ERROR=40, and CRITICAL=50. A level of
None or HIDDEN means that messages with this alias are always suppressed.
The following additional message categories are used widely by the RMS Node and are
defined by default:
* "`normal`" (default level 20=INFO) is used for any normal outcome.
* "`header`" (default level 20=INFO) is used for message headers following
:meth:`~PdsLogger.open` and summary messages following :meth:`~PdsLogger.close`.
* "`exception`" (default level 50=CRITICAL) is used when an exception is encountered.
* "`ds_store`" (default level 10=DEBUG) is used when a task encounters a ".DS_Store" file
as managed by the MacOS Finder.
* "`dot_`" (default level 40=ERROR) is used when a file beginning with "._" is
encountered. These files are sometimes created by "tar" commands in MacOS.
* "`invisible`" (default level 30=WARNING) is used if any other invisible file is
encountered.
Of course, any of these default levels can be modified on a logger-by-logger basis.
Use :meth:`~PdsLogger.set_limit` to specify a limit on the number of messages that can be
associated with an alias. For example, if the limit on "info" messages is 100, then log
messages after the hundredth will be suppressed, although the number of suppressed
messages will still be tracked. At the end of the log, a tally of the messages associated
with each alias is printed, including the number suppressed if the limit was exceeded.
:class:`PdsLogger` supports a rich set of formatting options for the log records, which
can be specified in the constructor or by using :meth:`~PdsLogger.set_format`. By default,
each log record automatically includes a time tag, log name, level, text message, and
optional file path. Also, use :meth:`~PdsLogger.add_root` and related methods to exercise
more control over how file paths appear.
Use :meth:`~PdsLogger.log` to log a message or else level-specific methods called
:meth:`~PdsLogger.debug`, :meth:`~PdsLogger.info`, :meth:`~PdsLogger.warning`, etc. Each
of these methods receives a message string and arguments identical to the methods of the
same name in `logging.Logger`. However, they take also an optional file path, which is
automatically formatted to appear after the message text in the log.
Method :meth:`~PdsLogger.exception` can be used inside a Python **except** clause to write
an exception message into the log, including the stacktrace. If you desire greater control
over how an exception is recorded in the log, you can use the :class:`LoggerException`
class or define your own subclass.
Simple tools are also provided to create handlers to assign to a :class:`PdsLogger` using
:meth:`~PdsLogger.add_handler` and related methods:
* :meth:`file_handler` is a function that provides a rich set of options for constructing
``logging.FileHandler`` objects, which allow logs to be written to a file. The options
include version numbering, appending a date or time to the file name, and daily
rotations. :meth:`file_handler` also supports the RMS Node's ``filecache`` module, which
allows log files to be seamlessly saved into cloud storage; simply pass in a URI or
``FCPath`` object instead of a local file path.
* :meth:`info_handler`, :meth:`~PdsLogger.warnings_handler`, and
:meth:`~PdsLogger.error_handler` are simpler versions of the above, in which the level
of message logging is implied.
* :meth:`stream_handler` is a function that creates handlers to write to an I/O stream
such as ``sys.stdout`` or ``sys.stderr``.
* ``STDOUT_HANDLER`` is a pre-defined handler that prints all output to the terminal.
* ``NULL_HANDLER`` is a pre-defined handler that suppresses all output.
Note that a :class:`PdsLogger` will print message to the terminal if no handler has been
assigned to it. As a result, if you really wish not to see any messages, you must assign
it the ``NULL_HANDLER``.
In the Macintosh Finder, log files are color-coded by the most severe message encountered
within the file: green for "info", yellow for warnings, red for errors, and violet for
critical or fatal errors.
For extremely simple logging needs, four subclasses of :class:`PdsLogger` are provided.
:class:`EasyLogger` prints all messages above a specified level of severity to the
terminal. :class:`ErrorLogger` only prints error messages. :class:`CriticalLogger` only
prints exceptions and other "critical" messages. :class:`NullLogger` suppresses all
messages, including logged exceptions. These four subclasses have the common trait that
they cannot be assigned handlers.
"""
import atexit
import datetime
import logging
import logging.handlers
import os
import re
import sys
import traceback
import warnings
from collections import defaultdict
from pathlib import Path
from filecache import FCPath
try:
import pdslogger._finder_colors as finder_colors
except ImportError: # pragma: no cover
# Exception is OK because finder_colors are not always used
pass
try:
from ._version import __version__
except ImportError: # pragma: no cover
__version__ = 'Version unspecified'
_TIME_FMT = '%Y-%m-%d %H:%M:%S.%f'
CRITICAL = logging.CRITICAL
FATAL = logging.CRITICAL
ERROR = logging.ERROR
WARNING = logging.WARNING
WARN = logging.WARNING
INFO = logging.INFO
DEBUG = logging.DEBUG
HIDDEN = 1 # Used for messages that are never displayed but might be summarized
_DEFAULT_LEVEL_NAME_ALIASES = {
'warn': 'warning',
'fatal': 'critical',
}
_DEFAULT_LEVEL_BY_NAME = {
# Standard level values
'critical': logging.CRITICAL, # 50
'error' : logging.ERROR, # 40
'warning' : logging.WARNING, # 30
'info' : logging.INFO, # 20
'debug' : logging.DEBUG, # 10
'hidden' : HIDDEN, # 1
# Additional level values defined for every PdsLogger
'normal' : logging.INFO,
'ds_store' : logging.DEBUG,
'dot_' : logging.ERROR,
'invisible': logging.WARNING,
'exception': logging.CRITICAL,
'header' : logging.INFO,
}
_DEFAULT_LEVEL_NAMES = {
logging.CRITICAL: 'critical', # 50
logging.ERROR : 'error', # 40
logging.WARNING : 'warning', # 30
logging.INFO : 'info', # 20
logging.DEBUG : 'debug', # 10
HIDDEN : 'hidden', # 1
}
_DEFAULT_LIMITS_BY_NAME = { # we're treating all as unlimited by default now
}
# Cache of names vs. PdsLoggers
_LOOKUP = {}
# Default global prefix, controlled by set_default_parent()
_DEFAULT_PARENT_NAME = 'pds'
_ATEXIT = False # changed to True during `atexit` cleanup
_BIGNUM = 2**63 - 1 # a very large integer
##########################################################################################
# PdsLogger class
##########################################################################################
[docs]
class PdsLogger(logging.Logger):
"""Logger class adapted by PDS Ring-Moon Systems Node.
See https://rms-pdslogger.readthedocs.io/en/latest/module.html for details.
"""
# Overridden by EasyLogger, etc.
_LOGGER_IS_FAKE = False # True to prevent registration
_NO_HANDLERS = False # True to prevent additional handlers
_NO_LEVELS = False # True to prevent changing the log level
# Overridden by as_pdslogger
_FROM_LOGGER = False
######################################################################################
# Constructors
######################################################################################
[docs]
def __init__(self, logname, *, parent=None, levels={}, limits={}, roots=[],
level=None, timestamps=None, digits=6, lognames=None, pid=False,
indent=None, levelnames=None, blanklines=True, colors=True, maxdepth=6):
"""Constructor for a PdsLogger.
Parameters:
logname (str or Logger):
Name of the logger, to be appended to the given `parent`. Each logger name
must be globally unique. If a Logger is given, its name is used and the
given Logger is wrapped by the returned PdsLogger.
parent (str, logger, or PdsLogger, optional)
The parent name or parent logger for this new logger. By default, the
logger defined by :meth:`~set_default_parent` is used. Use "" for no
parent.
levels (dict, optional):
A dictionary of level names and their values. These override or augment
the default level values. Use a negative level value to replace the
default name for a level; for example, {'VERY_BAD': -CRITICAL} will cause
"CRITICAL" errors to appear in the log as "VERY_BAD" errors instead. The
case of level names is ignored.
limits (dict, optional):
A dictionary indicating the upper limit on the number of messages to log
as a function of level name.
roots (str, Path, FCPath, or list[str, Path, or FCPath], optional):
Character strings to suppress if they appear at the beginning of file
paths. Used to reduce the length of log entries when, for example, every
file path is in a single subdirectory tree on the volume.
level (int or str, optional):
The minimum level or level name for a record to enter the log. By default,
this is the minimum logging level, HIDDEN + 1, unless this PdsLogger is
being derived from an existing Logger of the same name, in which case it
will match the existing Logger's level.
timestamps (bool, optional):
True to include timestamps in the log records. By default, this is True
unless this PdsLogger is being derived from an existing Logger of the same
name, in which case it is False.
digits (int, optional):
Number of fractional digits in the seconds field of the timestamp.
lognames (bool, optional):
True to include the name of the logger in the log records. By default,
this is True unless this PdsLogger is being derived from an existing
Logger of the same name, in which case it is False.
pid (bool, optional):
True to include the process ID in each log record.
indent (bool, optional):
True to include a sequence of dashes in each log record to provide a
visual indication of the depth in a logging hierarchy. By default, this is
True unless this PdsLogger is being derived from an existing Logger of the
same name, in which case it is False.
levelnames (bool, optional):
True to include the name of the log level, e.g., "ERROR", in the log
records. By default, this is True unless this PdsLogger is being derived
from an existing Logger of the same name, in which case it is False.
blanklines (bool, optional):
True to include a blank line in log files after a sub-logger is closed.
colors (bool, optional):
True to color-code any log files generated, for Macintosh only.
maxdepth (int, optional):
Maximum depth of the logging hierarchy, needed to prevent unlimited
recursion.
Raises:
ValueError: If the log name is already in use of if the level name associated
with a limit is unknown.
"""
# Handle a Logger as input
if isinstance(logname, PdsLogger):
raise ValueError(f'PdsLogger {logname.name} already exists')
if isinstance(logname, logging.Logger):
self._logger = logname
self._logname = self._logger.name
handlers = list(self._logger.handlers)
else:
# Expand the name
self._logname = PdsLogger._full_logname(logname, parent)
handlers = []
# Handle EasyLogger, etc.
if self._LOGGER_IS_FAKE:
self._logger = None
# Avoid duplicating the name of an existing PdsLogger
elif self._logname in _LOOKUP:
raise ValueError(f'PdsLogger {self._logname} already exists')
# If the Logger already exists, use it
elif self._logname in logging.Logger.manager.loggerDict:
self._logger = logging.getLogger(self._logname)
handlers = list(self._logger.handlers)
self._FROM_LOGGER = True
# ...and let it define the PdsLogger's default properties
level = self._logger.level if level is None else level
timestamps = False if timestamps is None else timestamps
lognames = False if lognames is None else lognames
indent = False if indent is None else indent
levelnames = False if levelnames is None else levelnames
# Otherwise, the underlying Logger is new
else:
self._logger = logging.getLogger(self._logname)
_LOOKUP[self._logname] = self
# Fill in the format parameters
self._timestamps = True if timestamps is None else timestamps
self._digits = digits
self._lognames = True if lognames is None else lognames
self._pid = os.getpid() if pid else 0
self._indent = True if indent is None else indent
self._levelnames = True if levelnames is None else levelnames
self._blanklines = bool(blanklines)
self._colors = bool(colors)
self._maxdepth = maxdepth
# Merge the dictionary of levels and their names
self._input_levels = levels
self._level_by_name = _DEFAULT_LEVEL_BY_NAME.copy() # name -> level
self._level_names = _DEFAULT_LEVEL_NAMES.copy() # level -> primary name
self._level_name_aliases = _DEFAULT_LEVEL_NAME_ALIASES.copy()
self._merge_level_names(levels)
# Define roots
roots = [roots] if isinstance(roots, str) else roots
self._roots = []
self.add_root(*roots)
# Support for multiple tiers in hierarchy
self._titles = [(self._logname, (), {}, True)] # (title, args, kwargs, logged)
self._start_times = [datetime.datetime.now()]
self._counters_by_name = [defaultdict(int)]
self._suppressed_by_name = [defaultdict(int)]
self._local_handlers = [[]] # handlers at this tier but not above
self._min_levels = [0]
# Fill in the level
level = HIDDEN + 1 if level is None else level
self.set_level(level)
self._input_limits = limits
self._limits_by_name = [defaultdict(lambda: _BIGNUM)] # pragma: no branch why??
self._limits_by_name[-1].update(_DEFAULT_LIMITS_BY_NAME)
for level_name, level_num in limits.items():
self.set_limit(level_name, level_num)
self._suppressions_logged = set() # level names having had a suppression message
# Mapping from log file absolute path to tuple (criticals, errors, warns)
self._log_file_summaries = {}
# Mappings from local file path to FCPath and FileHandler, if any
self._fcpath_by_local_abspath = {}
self._handler_by_local_abspath = {}
# Copy handlers from a pre-existing logger
self._handlers = [] # complete list of handlers across all tiers
self.add_handler(*handlers)
# Remove/close all handlers at program exit no matter what
atexit.register(self._remove_all_handlers_at_exit)
[docs]
@staticmethod
def get_logger(logname, *, parent=None, levels={}, limits={}, roots=[], level=None,
timestamps=None, digits=None, lognames=None, pid=None, indent=None,
levelnames=None, blanklines=None, colors=None, maxdepth=None):
"""Return the current logger by this name if it already exists; otherwise,
construct and return a new PdsLogger.
Parameters:
logname (str or Logger):
Name of the logger, to be appended to the given `parent`. If a Logger or
PdsLogger is provided, its name is used and `parent` is ignored.
parent (str, logger, or PdsLogger, optional)
The parent name or parent logger for this new logger. By default, the
logger defined by :meth:`~set_default_parent` is used. Use "" for no
parent.
levels (dict, optional):
A dictionary of level names and their values. These override or augment
the default level values.
limits (dict, optional):
A dictionary indicating the upper limit on the number of messages to log
as a function of level name.
roots (str, Path, FCPath, or list[str, Path, or FCPath], optional):
Character strings to suppress if they appear at the beginning of file
paths. Used to reduce the length of log entries when, for example, every
file path is in a single subdirectory tree on the volume.
level (int or str, optional):
The minimum level or level name for a record to enter the log. By default,
the value for an existing PdsLogger is preserved; if there is already a
Logger of the same name, the default is False; for an entirely new
PdsLogger, the default is the minimum logging level, HIDDEN + 1.
timestamps (bool, optional):
True to include timestamps in the log records. By default, the value for
an existing PdsLogger is preserved; if there is already a Logger of the
same name, the default is False; for an entirely new PdsLogger, the
default is True.
digits (int, optional):
Number of fractional digits in the seconds field of the timestamp. By
default, the value for an existing PdsLogger is preserved; otherwise, the
default is 6.
lognames (bool, optional):
True to include the name of the logger in the log records. By default, the
value for an existing PdsLogger is preserved; if there is already a Logger
of the same name, the default is False; for an entirely new PdsLogger, the
default is True.
pid (bool, optional):
True to include the process ID in each log record. By default, the value
for an existing PdsLogger is preserved; otherwise, the default is False.
indent (bool, optional):
True to include a sequence of dashes in each log record to provide a
visual indication of the depth in a logging hierarchy. By default, the
value for an existing PdsLogger is preserved; if there is already a Logger
of the same name, the default is False; for an entirely new PdsLogger, the
default is True.
levelnames (bool, optional):
True to include the name of the log level, e.g., "ERROR", in the log
records. By default, the value for an existing PdsLogger is preserved; if
there is already a Logger of the same name, the default is False; for an
entirely new PdsLogger, the default is True.
blanklines (bool, optional):
True to include a blank line in log files when a tier in the hierarchy is
closed; False otherwise. By default, the value for an existing PdsLogger
is preserved; otherwise, the default is True.
colors (bool, optional):
True to color-code any log files generated, for Macintosh only. By
default, the value for an existing PdsLogger is preserved; otherwise, the
default is True.
maxdepth (int, optional):
Maximum depth of the logging hierarchy, needed to prevent unlimited
recursion. By default, the value for an existing PdsLogger is preserved;
otherwise, the default is 6.
"""
if isinstance(logname, logging.Logger):
logname = logname.name
else:
logname = PdsLogger._full_logname(logname, parent)
if logname in _LOOKUP:
plogger = _LOOKUP[logname]
plogger.set_format(level=level, timestamps=timestamps, digits=digits,
lognames=lognames, pid=pid, indent=indent,
levelnames=levelnames, blanklines=blanklines,
colors=colors, maxdepth=maxdepth)
plogger._merge_level_names(levels)
for name, value in limits.items():
plogger.set_limit(name, value)
roots = [roots] if isinstance(roots, str) else roots
plogger.add_root(*roots)
return plogger
# Set defaults depending on whether the Logger already exists
digits = 6 if digits is None else digits
pid = False if pid is None else pid
blanklines = True if blanklines is None else blanklines
colors = True if colors is None else colors
maxdepth = 6 if maxdepth is None else maxdepth
if logname in logging.Logger.manager.loggerDict:
timestamps = False if timestamps is None else timestamps
lognames = False if lognames is None else lognames
indent = False if indent is None else indent
levelnames = False if indent is None else indent
else:
timestamps = True if timestamps is None else timestamps
lognames = True if lognames is None else lognames
indent = True if indent is None else indent
levelnames = True if indent is None else indent
return PdsLogger(logname, parent=parent, levels=levels, limits=limits,
roots=roots, level=level, timestamps=timestamps, digits=digits,
lognames=lognames, pid=pid, indent=indent, levelnames=levelnames,
blanklines=blanklines, colors=colors, maxdepth=maxdepth)
[docs]
@staticmethod
def getLogger(*args, **kwargs):
"""Alternative name for get_logger()."""
return PdsLogger.get_logger(*args, **kwargs)
[docs]
def copy(self, logname, parent=None):
"""A copy of this PdsLogger with a revised name.
All other attributes are retained except handlers.
Parameters:
logname (str):
Name of the logger, to be appended to the given `parent`.
parent (str, logger, or PdsLogger, optional)
The parent name or parent logger for this new logger. By default, the
logger defined by :meth:`~set_default_parent` is used. Use "" for no
parent.
Raises:
ValueError: If the log name is already in use of if the level name associated
with a limit is unknown.
"""
return type(self)(logname, parent=parent, levels=self._input_levels,
limits=self._input_limits, roots=self._roots, level=self.level,
timestamps=self._timestamps, digits=self._digits,
lognames=self._lognames, pid=self._pid, indent=self._indent,
levelnames=self._levelnames, blanklines=self._blanklines,
colors=self._colors, maxdepth=self._maxdepth)
[docs]
def get_child(self, name, levels={}, limits={}, roots=[], level=None,
timestamps=None, digits=None, lognames=None, pid=None, indent=None,
levelnames=None, blanklines=None, colors=None, maxdepth=None):
"""Construct if necessary and return the named child PdsLogger of this PdsLogger.
Parameters:
name (str):
Name of the child logger, to be appended to the name of this PdsLogger.
levels (dict, optional):
A dictionary of level names and their values. These values override or
augment the level values of the parent logger.
limits (dict, optional):
A dictionary indicating the upper limit on the number of messages to log
as a function of level name. These values override or augment the limits
of the parent logger.
roots (str, Path, FCPath, or list[str, Path, or FCPath], optional):
Character strings to suppress if they appear at the beginning of file
paths. Used to reduce the length of log entries when, for example, every
file path is in a single subdirectory tree on the volume. These values
augment the roots inherited from the parent logger.
level (int or str, optional):
The minimum level or level name for a record to enter the log. If not
specified, the child logger inherits its value from the parent.
timestamps (bool, optional):
True to include timestamps in the log records. If not specified, the child
logger inherits this value from the parent.
digits (int, optional):
Number of fractional digits in the seconds field of the timestamp. If not
specified, the child logger inherits this value from the parent.
lognames (bool, optional):
True to include the name of the logger in the log records. If not
specified, the child logger inherits this value from the parent.
pid (bool, optional):
True to include the process ID in each log record. If not specified, the
child logger inherits this value from the parent.
indent (bool, optional):
True to include a sequence of dashes in each log record to provide a
visual indication of the depth in a logging hierarchy. If not specified,
the child logger inherits this value from the parent.
levelnames (bool, optional):
True to include the name of the log level, e.g., "ERROR", in the log
records. If not specified, the child logger inherits this value from the
parent.
blanklines (bool, optional):
True to include a blank line in log files when a tier in the hierarchy is
closed; False otherwise. If not specified, the child logger inherits this
value from the parent.
colors (bool, optional):
True to color-code any log files generated, for Macintosh only. If not
specified, the child logger inherits this value from the parent.
maxdepth (int, optional):
Maximum depth of the logging hierarchy, needed to prevent unlimited
recursion. If not specified, the child logger inherits this value from the
parent.
"""
child = self.copy(name, parent=self)
child._merge_level_names(levels)
for name, value in limits.items():
child.set_limit(name, value)
roots = [roots] if isinstance(roots, str) else roots
child.add_root(*roots)
child.set_format(level=level, timestamps=timestamps, digits=digits,
lognames=lognames, pid=pid, indent=indent, levelnames=levelnames,
blanklines=blanklines, colors=colors, maxdepth=maxdepth)
return child
[docs]
def getChild(self, name, **kwargs):
"""Alternative name for get_child()."""
return self.get_child(name, **kwargs)
[docs]
@staticmethod
def as_pdslogger(logger):
"""Convert the given Logger to a PdsLogger.
Parameters:
logger (logging.Logger or PdsLogger): logger to convert to PdsLogger.
Returns:
PdsLogger: converted logger. A PdsLogger is returned as is. If a PdsLogger has
already been defined for the given logger, that PdsLogger is returned.
Otherwise, a new PdsLogger is constructed using properties that will match the
Logger's current format.
"""
if isinstance(logger, PdsLogger):
return logger
if logger.name in _LOOKUP:
return _LOOKUP[logger.name]
new_logger = PdsLogger.get_logger(logger.name, parent='', level=logger.level,
timestamps=False, lognames=False, pid=False,
indent=False, levelnames=False,
blanklines=False)
new_logger.add_handler(*logger.handlers)
new_logger._FROM_LOGGER = True
return new_logger
[docs]
def __str__(self):
level_name = self._level_names.get(self.level, str(self.level)).upper()
return f'<{type(self).__name__} {self.name} (Level {level_name})>'
[docs]
def __repr__(self):
return str(self)
######################################################################################
# Formatting support
######################################################################################
######################################################################################
# Name and parent/child support
######################################################################################
@property
def name(self):
"""The name of this PdsLogger."""
return self._logname
@property
def parent(self):
"""The parent of this PdsLogger."""
if self._logger is None or self._logger.parent is None:
return None
return PdsLogger.as_pdslogger(self._logger.parent)
[docs]
@staticmethod
def set_default_parent(parent):
"""Define the name of the default prefix, which will be used for each new logger
unless overridden in the constructor or in the call to get_logger().
Parameters:
parent (str, logger, or PdsLogger): Name or logger to serve as the default
parent.
"""
global _DEFAULT_PARENT_NAME
if isinstance(parent, logging.Logger):
_DEFAULT_PARENT_NAME = parent.name
else:
_DEFAULT_PARENT_NAME = parent.rstrip('.')
if sys.version_info >= (3, 12): # pragma: no cover
[docs]
def get_children(self):
"""The set of child PdsLoggers of this PdsLogger."""
children = self._logger.getChildren()
return {PdsLogger.as_pdslogger(child) for child in children}
[docs]
def getChildren(self):
"""Alternative name for get_children()."""
return self.get_children()
@staticmethod
def _full_logname(logname, parent=None):
"""The full log name with the default parent pre-pended if necessary.
If the logname matches any part of the parent, only the prior part of the parent
is pre-pended, so if parent = 'a.b' and logname = 'b.c', then the result is
'a.b.c'.
"""
if parent is None:
parent = _DEFAULT_PARENT_NAME
if not parent:
return logname if logname else 'root'
if isinstance(parent, logging.Logger):
return parent.name + '.' + logname
parent = parent.rstrip('.')
parent_parts = parent.split('.')
logname_parts = logname.split('.')
try:
indx = parent_parts.index(logname_parts[0])
except ValueError:
return parent + '.' + logname
return '.'.join(parent_parts[:indx] + logname_parts)
######################################################################################
# Level support
######################################################################################
@property
def level(self):
"""The minimum logging level of this PdsLogger."""
return self._min_levels[-1]
[docs]
def set_level(self, level):
"""Set the level of messages for the current tier in the logger's hierarchy.
Parameters;
level (int or str, optional):
The minimum level or level name for a record to enter the log.
"""
if self._NO_LEVELS: # for ErrorLogger, CriticalLogger, etc.
return
if isinstance(level, str):
self._min_levels[-1] = self._level_by_name[self._repair_level_name(level)]
else:
self._min_levels[-1] = level
if self._logger:
self._logger.setLevel(self._min_levels[-1])
[docs]
def setLevel(self, level):
"""Alternative name for set_level()."""
self.set_level(level)
def _merge_level_names(self, levels):
"""Merge the given dictionary mapping level names to numbers into the internal
dictionaries.
"""
for level_name, level_num in levels.items():
level_name = level_name.lower()
# If negative, define this name as the default
if level_num < 0:
level_num = -level_num
old_level_name = self._level_names.get(level_num, '')
self._level_names[level_num] = level_name
# Update alias lookup...
for name, alias in list(self._level_name_aliases.items()):
if name == level_name: # remove new name as an alias
del self._level_name_aliases[name]
elif old_level_name == alias: # point aliases of old name to new
self._level_name_aliases[name] = level_name
# Alias old name to new name
if old_level_name and old_level_name != level_name:
self._level_name_aliases[old_level_name] = level_name
# Add new name to the name lookup dictionary
self._level_by_name[level_name] = level_num
# Add mapping from level number to name if it's unique
if level_num not in self._level_names:
self._level_names[level_num] = level_name
# Apply these level names to the parent logger to support propagation
# This will be automatically carried all the way to the root logger via recursion.
if self._logger is not None and self._logname != 'root':
logname = self._logname.rpartition('.')[0]
PdsLogger.get_logger(logname, parent='', levels=self._input_levels)
######################################################################################
# Limit support
######################################################################################
[docs]
def set_limit(self, name, limit):
"""Set the upper limit on the number of messages with this level name.
Parameters:
name (int or str): A logging level or level name.
limit (int): The maximum number of messages to be logged at this level. A
limit of -1 implies no limit.
"""
name = self._repair_level_name(name)
if name not in self._level_by_name:
raise ValueError('undefined level name: ' + repr(name))
self._limits_by_name[-1][name] = limit if limit >= 0 else _BIGNUM
[docs]
def get_limit(self, name):
"""Get the current upper limit on the number of messages with this level name.
Parameters:
name (int or str): A logging level or level name.
Raises:
KeyError: If `name` is not a known level name.
"""
name = self._repair_level_name(name)
if name not in self._level_by_name:
raise KeyError('undefined level name: ' + repr(name))
return self._limits_by_name[-1][name]
[docs]
def get_limits(self):
"""A dictionary of the current default upper limit on the number of messages with
each level name.
"""
new_dict = {}
for name in self._level_by_name:
new_dict[name] = self._limits_by_name[-1][name]
return new_dict
######################################################################################
# Root support
######################################################################################
[docs]
def add_root(self, *roots):
"""Add one or more paths to the set of root paths.
When a root path appears at the beginning of a logged file path, only the portion
following the root is shown.
Parameters:
*roots (str, Path, or FCPath): One or more root paths to suppress.
"""
root_list = [] # there is a case in pdsfile where the input is a list
for root in roots:
if isinstance(root, (list, tuple, set)):
root_list += list(root)
else:
root_list.append(root)
for root in root_list:
root = str(root).rstrip('/') + '/'
if root not in self._roots:
self._roots.append(root)
self._roots.sort(key=lambda x: (-len(x), x)) # longest patterns first
@property
def roots(self):
"""This current list of root paths as strings."""
return list(self._roots)
[docs]
def replace_root(self, *roots):
"""Replace the existing root(s) with one or more new paths."""
self._roots = []
self.add_root(*roots)
######################################################################################
# Handler API
######################################################################################
@property
def handlers(self):
"""The list of handlers of this PdsLogger."""
return self._handlers
[docs]
def has_handlers(self):
"""True if this PdsLogger has handlers."""
return bool(self._handlers)
[docs]
def hasHandlers(self):
"""Alternative name for has_handlers."""
return self.has_handlers()
[docs]
def add_handler(self, *handlers, local=None):
"""Add one or more handlers to this PdsLogger at the current location in the
hierarchy.
If a handler is already in use, or if it refers to a log file that is already in
use, the input is ignored silently.
Parameters:
*handlers (logging.Handler, str, path, or FCPath):
One or more handlers. If a file path is provided, it is used by
:meth:`file_handler` to construct a new FileHandler.
local (bool, optional):
True to add this as a local handler, which will be removed when
:meth:`~PdsLogger.close` is called; False to treat it as a global handler,
which is preserved until explicitly removed. Outside any call to
:meth:`~PdsLogger.open`, all handlers are global.
"""
# EasyLogger and its subclasses do not accept handlers; warn
if self._NO_HANDLERS:
if handlers and not hasattr(self, '_warned_about_handlers'):
warnings.warn(f'class {type(self).__name__} does not accept handlers')
self._warned_about_handlers = True
return
if local is None:
local = self._get_depth() > 0
# Add each new handler if its absolute path is unique
for handler in list(handlers): # work from a copy just in case
if handler in self._handlers: # no duplicate handlers
continue
# Convert a file path to a handler
if isinstance(handler, (str, Path, FCPath)):
handler = file_handler(handler)
# Save a FileHandler in the global dictionary by absolute path string
if isinstance(handler, logging.FileHandler):
abspath = str(Path(handler.baseFilename).expanduser()
.absolute().resolve())
if abspath in self._handler_by_local_abspath:
continue # log file already in use
self._handler_by_local_abspath[abspath] = handler
self._log_file_summaries[abspath] = self.summarize(local=local)
# Save a remote FCPath based on retrieved name so we can close it when the
# handler is removed
if hasattr(handler, 'fcpath') and not handler.fcpath.is_local():
self._fcpath_by_local_abspath[abspath] = handler.fcpath
self._handlers.append(handler)
self._local_handlers[-1 if local else 0].append(handler)
self._logger.addHandler(handler)
[docs]
def addHandler(self, *handlers, local=False):
"""Alternative name for add_handler()."""
self.add_handler(*handlers)
[docs]
def remove_handler(self, *handlers):
"""Remove one or more handlers from this PdsLogger.
Parameters:
*handlers (logging.Handler, str, path, FCPath):
One or more handlers. If a file path is provided, the FileHandler using
that log path is removed.
"""
def remove_one_handler(handler):
"""Remove handler if found; otherwise, pass."""
self._logger.removeHandler(handler)
try:
self._handlers.remove(handler)
except ValueError:
return
for local_list in self._local_handlers:
if handler in local_list:
local_list.remove(handler)
if self._NO_HANDLERS: # if EasyLogger, NullLogger, etc.
return
for handler in handlers:
# Identify handler and path if any
if isinstance(handler, (str, Path, FCPath)):
path = handler
handler = None
elif isinstance(handler, logging.FileHandler):
path = handler.baseFilename
else:
path = ''
# Without a file path, this is easy
if not path:
remove_one_handler(handler)
continue
# Identify an FCPath and abspath if any
if hasattr(handler, 'fcpath'):
fcpath = handler.fcpath
abspath = str(fcpath.get_local_path().expanduser().absolute().resolve())
elif isinstance(path, FCPath):
fcpath = path
abspath = str(fcpath.get_local_path().expanduser().absolute().resolve())
else:
abspath = str(Path(path).expanduser().absolute().resolve())
fcpath = self._fcpath_by_local_abspath.get(abspath, None)
# Check an abspath against the list of handlers
handler = self._handler_by_local_abspath.get(abspath, None)
if not handler:
continue # abspath not in use
# Manage the files
local = handler not in self._local_handlers[0]
new_summary = self.summarize(local=local)
old_summary = self._log_file_summaries[abspath]
# Remove the handler from the lists
remove_one_handler(handler)
# Remove the abspath from the dictionaries
del self._handler_by_local_abspath[abspath]
del self._log_file_summaries[abspath]
if abspath in self._fcpath_by_local_abspath:
del self._fcpath_by_local_abspath[abspath]
# During `atexit` call, the file may already have been cleaned up
if os.path.exists(abspath): # pragma: no branch
# If it's an FCPath, issue an explicit upload
if fcpath:
try:
fcpath.upload()
except NotImplementedError:
warnings.warn(f'remote file "{fcpath}" cannot be uploaded; '
f'local path is "{abspath}"')
if self._colors: # pragma: no cover
# If the xattr module has been imported on a Mac, set the color of
# the log file to indicate outcome.
try:
if new_summary[0] - old_summary[0] > 0:
finder_colors.set_color(abspath, 'violet')
elif new_summary[1] - old_summary[1] > 0:
finder_colors.set_color(abspath, 'red')
elif new_summary[2] - old_summary[2] > 0:
finder_colors.set_color(abspath, 'yellow')
else:
finder_colors.set_color(abspath, 'green')
except (AttributeError, NameError):
pass
[docs]
def removeHandler(self, *handlers):
"""Alternative name for remove_handler()."""
self.remove_handler(*handlers)
[docs]
def remove_all_handlers(self):
"""Remove all the handlers from this PdsLogger."""
for handler in list(self._handlers): # use a copy of the list; it's changing
self.remove_handler(handler)
def _remove_all_handlers_at_exit(self):
"""Remove all the handlers from this PdsLogger at program exit.
This ensures that if a program aborts, a remote log is still uploaded to the
cloud.
"""
global _ATEXIT
_ATEXIT = True # pragma: no cover (executes after tests end)
self.remove_all_handlers() # pragma: no cover
[docs]
def replace_handler(self, *handlers):
"""Replace the existing local handlers with one or more new handlers at the
current tier in the logging hierarchy.
Parameters:
*handlers (str, path, FCPath, or logging.Handler):
One or more handlers. If a file path is provided, it is used in a new
FileHandler.
"""
self.remove_handler(*list(self._local_handlers[-1]))
self.add_handler(*handlers)
######################################################################################
# More logger.Logging API
######################################################################################
@property
def propagate(self):
return self._logger.propagate if self._logger else False # False for EasyLogger
@propagate.setter
def propagate(self, value):
if self._logger:
self._logger.propagate = bool(value)
@property
def disabled(self):
return self._logger.disabled if self._logger else False
[docs]
def isEnabledFor(self, level):
return self._logger.isEnabledFor(level) if self._logger else (level >= self.level)
[docs]
def getEffectiveLevel(self):
return self._logger.getEffectiveLevel() if self._logger else self.level
[docs]
def addFilter(self, filter):
if self._logger: # pragma: no cover
self._logger.addFilter(filter)
[docs]
def removeFilter(self, filter):
if self._logger: # pragma: no cover
self._logger.removeFilter(filter)
[docs]
def filter(self, record):
return self._logger.filter(record) if self._logger else True # pragma: no cover
[docs]
def findCaller(self, stack_info=False, stacklevel=1):
if self._logger: # pragma: no cover
return self._logger.findCaller(stack_info=stack_info, stacklevel=stacklevel)
return None # pragma: no cover
[docs]
def handle(self, record):
if self._logger: # pragma: no cover
self._logger.handle(record)
[docs]
def makeRecord(self, *args, **kwargs):
if self._logger: # pragma: no cover
return self._logger.makeRecord(*args, **kwargs)
return None # pragma: no cover
# Alternative names
is_enabled_for = isEnabledFor
get_effective_level = getEffectiveLevel
add_filter = addFilter
remove_filter = removeFilter
make_record = makeRecord
######################################################################################
# Open/close
######################################################################################
class _Closer():
"""Context Manager used by open()."""
def __init__(self, logger):
self.logger = logger
def __enter__(self):
pass
def __exit__(self, exc_type, exc_val, exc_tb):
self.logger.close()
[docs]
def open(self, title, *args, filepath='', level=None, limits={}, handler=[],
force=False, blankline=False, **kwargs):
"""Begin a new tier in the logging hierarchy.
Parameters:
title (str):
Title of the new section of the log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns inside
the message string (indicated by "%" of "{") that use these args, a single
argument is interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Optional file path to include in the title.
level (int, or str, optional):
The level or level name for the minimum logging level to use within this
tier. The default is to preserve the current logging level.
limits (dict, optional):
A dictionary mapping level name to the maximum number of messages of that
level to include. Subsequent messages are suppressed. Use a limit of -1 to
show all messages.
handler (Handler or list[Handler], optional):
Optional handler(s) to use only until this part of the logger is closed.
force (bool, optional):
True to force the logging of the "open" messages, even if the current
logging level is above that specified by `level`.
blankline (bool, optional):
True to insert a blank line before the logger.
**kwargs:
Zero or more keyword=value attributes to be substituted into the title
string using the string formatted operator.
Returns:
context manager:
A context manager enabling "`open logger.open():`" syntax. If `open()` is
not being used as a context manager, this object can be ignored.
"""
# Check the hierarchy depth
if self._get_depth() >= self._maxdepth:
raise ValueError('Maximum logging hierarchy depth has been reached')
# Format the title + filepath
if not filepath and len(args) == 1 and not _message_uses_args(title):
filepath = args[0]
args = []
header_level = self._level_by_name['header']
title = self._format_message(header_level, title, *args, **kwargs)
if filepath:
title += ': ' + self._logged_filepath(filepath)
# Determine the new logging level
if level is None:
new_level = self._min_levels[-1]
elif isinstance(level, str):
new_level = self._level_by_name[self._repair_level_name(level)]
else:
new_level = level
# Determine whether to log the header
header_logged = (header_level >= min(self._min_levels[-1], new_level)) or force
# Update the tier-specific info
self._min_levels.append(new_level)
self.set_level(new_level)
self._start_times.append(datetime.datetime.now())
self._local_handlers.append([])
self._titles.append((title, args, kwargs, header_logged))
# Set the level-specific limits
self._limits_by_name.append(defaultdict(int))
for name, limit in limits.items():
self._limits_by_name[-1][name] = limit if limit >= 0 else _BIGNUM
# Unless overridden, each tier is bound by the limits of the tier above
for name, limit in self._limits_by_name[-2].items():
if name not in self._limits_by_name[-1] and limit < _BIGNUM:
count_so_far = sum(dict_[name] for dict_ in self._counters_by_name)
new_limit = max(0, limit - count_so_far)
self._limits_by_name[-1][name] = new_limit
# Create new message counters for this tier
self._counters_by_name.append(defaultdict(int))
self._suppressed_by_name.append(defaultdict(int))
# Update the handlers
if handler:
handlers = handler if isinstance(handler, (list, tuple)) else [handler]
self.add_handler(*handlers, local=True)
# Write header message
if header_logged:
if blankline:
self.blankline(header_level)
self._logger_log(header_level, self._logged_text('HEADER', title, shift=-1))
# For use of open() as a context manager
return PdsLogger._Closer(self)
[docs]
def summarize(self, local=True):
"""Return a tuple describing the number of logged messages by category in the
current tier of the hierarchy.
Parameters:
local (bool, optional): True for the totals at this depth in the hierarchy;
False for the global totals since this PdsLogger was created.
Returns:
tuple: (number of critical errors, number of errors, number of warnings, total
number of messages). These counts include messages that were suppressed
because a limit was reached.
"""
criticals = 0
errors = 0
warnings = 0
total = 0
indices = (-1,) if local else tuple(range(len(self._counters_by_name)))
for indx in indices:
for dict_ in (self._counters_by_name[indx], self._suppressed_by_name[indx]):
for name, count in dict_.items():
level = self._level_by_name[name]
if level >= CRITICAL:
criticals += count
elif level >= ERROR:
errors += count
elif level >= WARNING:
warnings += count
total += count
return (criticals, errors, warnings, total)
[docs]
def close(self, *, force=False, blankline=False):
"""Close the log at its current depth in the hierarchy.
The closure is logged, plus a summary of the time elapsed and levels identified
while this sub-logger was open.
Parameters:
force (bool, int, or str, optional):
True to force the logging of all summary messages. Alternatively use a
level or level name to force the summary messages only about logged
messages at this level and higher.
blankline (bool, optional):
True to insert a blank line in the log before closing if it would not
otherwise be inserted.
Returns:
tuple: (number of critical errors, number of errors, number of warnings, total
number of messages). These counts include messages that were suppressed
because a limit was reached.
"""
# Interpret the `force` input
if isinstance(force, str):
min_level_for_log = self._level_by_name[self._repair_level_name(force)]
elif not isinstance(force, bool):
min_level_for_log = force
elif force:
min_level_for_log = HIDDEN + 1
elif len(self._min_levels) >= 2:
min_level_for_log = min(self._min_levels[-2:])
else:
min_level_for_log = self._min_levels[-1]
# Create a list of messages summarizing results; each item is (logged level, text)
header_level = self._level_by_name['header']
(title, title_args, title_kwargs, header_logged) = self._titles[-1]
messages = []
if header_logged: # log the summary only if the header was logged
if self._timestamps:
elapsed = datetime.datetime.now() - self._start_times[-1]
messages += [(header_level, 'Elapsed time = ' + str(elapsed))]
# Define messages indicating counts by level name
tuples = [(level, name) for name, level in self._level_by_name.items()]
tuples.sort(reverse=True)
for level, name in tuples:
count = self._counters_by_name[-1][name]
suppressed = self._suppressed_by_name[-1][name]
if count + suppressed == 0:
continue
capname = name.upper()
if suppressed == 0:
plural = '' if count == 1 else 's'
note = f'{count} {capname} message{plural}'
else:
plural = '' if count == 1 else 's'
note = (f'{count} {capname} message{plural} reported of '
f'{count + suppressed} total')
messages += [(level, note)]
# Transfer the totals to the logger tier above
if len(self._counters_by_name) > 1:
for name, count in self._counters_by_name[-1].items():
self._counters_by_name[-2][name] += count
self._suppressed_by_name[-2][name] += self._suppressed_by_name[-1][name]
# Determine values to return
(criticals, errors, warnings, total) = self.summarize(local=True)
# Log the summary
if header_logged:
self._logger_log(header_level,
self._logged_text('SUMMARY', 'Completed: ' + title,
shift=-1),
*title_args, **title_kwargs)
for level, note in messages:
if level >= min_level_for_log:
self._logger_log(header_level, self._logged_text('SUMMARY', note,
shift=-1))
if blankline or self._blanklines:
self.blankline(header_level)
# Remove any handlers at this tier
self.remove_handler(*self._local_handlers[-1])
# Back up one level in the hierarchy
if len(self._titles) > 1:
self._titles = self._titles[:-1]
self._start_times = self._start_times[:-1]
self._limits_by_name = self._limits_by_name[:-1]
self._counters_by_name = self._counters_by_name[:-1]
self._suppressed_by_name = self._suppressed_by_name[:-1]
self._local_handlers = self._local_handlers[:-1]
self._min_levels = self._min_levels[:-1]
else: # closing the outermost tier is a reset
self._start_times = [datetime.datetime.now()]
self._counters_by_name = [defaultdict(int)]
self._suppressed_by_name = [defaultdict(int)]
if self._logger and self._min_levels:
self._logger.setLevel(self._min_levels[-1])
return (criticals, errors, warnings, total)
[docs]
def message_count(self, name):
"""The number of messages generated at this named level since this last open().
Parameters:
name (str): Name of a level.
Returns:
int: The number of messages logged, including any suppressed if a limit was
reached.
"""
name = self._repair_level_name(name)
return self._counters_by_name[-1][name] + self._suppressed_by_name[-1][name]
######################################################################################
# Logging methods
######################################################################################
[docs]
def log(self, level, message, *args, filepath='', force=False, suppress=False,
**kwargs):
"""Log one record.
Parameters:
level (int or str):
Logging level or level name.
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns inside
the message string (indicated by "%" of "{") that use these args, a single
argument is interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
# Interpret the args
if not filepath and len(args) == 1 and not _message_uses_args(message):
filepath = args[0]
args = []
# Determine the level name and number
if isinstance(level, str):
level_name_for_log = self._repair_level_name(level)
level_name_for_count = level_name_for_log
level_for_log = self._level_by_name[level_name_for_log]
elif level in self._level_names:
level_name_for_log = self._level_names[level]
level_name_for_count = level_name_for_log
level_for_log = level
else: # Level is not one of 10, 20, 30, etc.
level_for_log = level
level_name_for_log = self._logged_level_name(level) # e.g., "ERROR+1"
level_for_count = max(10*(level//10), HIDDEN)
level_name_for_count = self._level_names[level_for_count]
# Get the count and limit for messages with this level name
count = self._counters_by_name[-1][level_name_for_count]
limit = self._limits_by_name[-1].get(level_name_for_count, -1)
# Determine whether to print
if isinstance(force, str):
force = self._level_by_name[self._repair_level_name(force)]
forced = False
if not isinstance(force, bool):
level_for_log = max(force, level_for_log)
log_now = level_for_log >= self._min_levels[-1]
forced = log_now
elif force:
log_now = level_for_log >= self._min_levels[-1]
forced = log_now and limit >= 0 and count >= limit
elif suppress:
log_now = False
elif level_for_log < self._min_levels[-1]:
log_now = False
elif limit < 0: # -1 means no limit
log_now = True
elif count >= limit:
log_now = False
else:
log_now = True
# Log now
if log_now:
text = self._logged_text(level_name_for_log, message, filepath)
self._logger_log(level_for_log, text, *args, **kwargs)
self._increment_count(level_name_for_count, suppressed=False)
# Log a suppression message next time this level is suppressed
if not forced:
self._suppressions_logged.discard(level_name_for_count)
# Otherwise...
else:
self._increment_count(level_name_for_count, suppressed=True)
# If this is the first suppressed message due to the limit, notify
if (not suppress and limit >= 0
and level_for_log >= self._min_levels[-1]
and level_name_for_count not in self._suppressions_logged):
message = f'Additional {level_name_for_count.upper()} messages suppressed'
text = self._logged_text(level_name_for_count, message)
self._logger_log(level_for_log, text)
self._suppressions_logged.add(level_name_for_count)
def _increment_count(self, name, suppressed=False):
"""Increment the count of logged messages here and in the propagated parents."""
plogger = self
while plogger is not None:
if suppressed:
plogger._suppressed_by_name[-1][name] += 1
else:
plogger._counters_by_name[-1][name] += 1
if not plogger.propagate:
break
plogger = plogger.parent
[docs]
def debug(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "debug".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('debug', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def info(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "info".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('info', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def warning(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "warning".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('warn', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def warn(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "warning".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('warn', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def error(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "error".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('error', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def critical(self, message, *args, filepath='', force=False, suppress=False,
**kwargs):
"""Log a message with level == "critical".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('critical', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def fatal(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "fatal", equivalent to "critical".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('fatal', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def normal(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "normal", equivalent to "info".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('normal', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def ds_store(self, message, *args, filepath='', force=False, suppress=False,
**kwargs):
"""Log a message with level == "ds_store", indicating that a file named
".DS_Store" was found.
These files are sometimes created on a Mac.
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('ds_store', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def dot_underscore(self, message, *args, filepath='', force=False, suppress=False,
**kwargs):
"""Log a message with level == `"dot_"`, indicating that a file with a name
beginning with "._" was found.
These files are sometimes created during file transfers from a Mac.
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('dot_', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def invisible(self, message, *args, filepath='', force=False, suppress=False,
**kwargs):
"""Log a message with level == "invisible", indicating that an invisible file was
found.
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('invisible', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def hidden(self, message, *args, filepath='', force=False, suppress=False, **kwargs):
"""Log a message with level == "hidden".
Parameters:
message (str):
Message to log.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
Path of the relevant file, if any.
force (bool, int, or str, optional):
True to log this message even if the relevant limit has been reached;
False otherwise. Alternatively, use a level or level name to log the
given message to handlers at or below a particular level.
suppress (bool, optional):
True to suppress message reporting even if the relevant limit has not been
reached. The message is still included in the count. The `force` option
takes precedence over this option.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
"""
self.log('hidden', message, *args, filepath=filepath, force=force,
suppress=suppress, **kwargs)
[docs]
def exception(self, error, *args, filepath='', stacktrace=None, exc_info=None,
more='', **kwargs):
"""Log an Exception or KeyboardInterrupt.
This method is only to be used inside an `except` clause.
Parameters:
error (Exception or str):
The error raised or a text message.
*args:
Zero or more items to be substituted into the message string using the
string formatting operator. If there are no substitution patterns
(indicated by "%" of "{") inside the message string, a single argument is
interpreted as the `filepath`.
filepath (str, Path, or FCPath, optional):
File path to include in the message.
stacktrace (bool, optional):
True to include the stacktrace of the exception.
exc_info (bool, optional):
Alternative name for `stacktrace`. If both are unspecified, True is
assumed.
more (str, optional):
Additional information to write into the log following the error message.
**kwargs:
Zero or more keyword=value attributes to be substituted into the message
string using the string formatted operator.
Notes:
The :class:`LoggerError` constructor provides additional contol over how an
exception will appear in the log, so use it or a subclass if you need this
extra control.
A KeyboardInterrupt is not handled by this method. A brief note is added to
the log and then this exception is re-raised.
"""
# Get the error status
etype, sys_error, tb = sys.exc_info()
if etype is None: # pragma: no cover (can't simulate)
return # Exception was already handled
# KeyboardInterrupt is a special case
if isinstance(sys_error, KeyboardInterrupt): # pragma: no cover (can't simulate)
self.fatal('**** Interrupted by user')
raise sys_error
# Get stacktrace status
if stacktrace is None:
stacktrace = exc_info
if stacktrace is None:
stacktrace = True
# Interpret the input type
if isinstance(error, str):
message = error
error = sys_error
else:
message = ''
etype = type(error)
# LoggerError is a special case
if isinstance(error, LoggerError):
if isinstance(error.level, str):
level = self._level_by_name[error.level]
else:
level = error.level
message = self._format_message(level, message or str(error), *args, **kwargs)
self.log(level, message, filepath=filepath, force=error.force)
if more:
self._logger_log(level, more)
if stacktrace:
tb = error.tb if error.tb else tb
self._logger_log(level, ''.join(traceback.format_tb(tb)))
return
# Any other Exception...
if not message:
message = '**** ' + etype.__name__ + ' ' + str(error)
if not filepath and len(args) == 1 and not _message_uses_args(message):
filepath = args[0]
args = []
# Use logger.exception if this PdsLogger was derived from a logging.Logger
if self._FROM_LOGGER:
message = self._format_message(ERROR, message, *args, **kwargs)
if filepath:
message += ': ' + self._logged_filepath(filepath)
self._logger.exception(message, *args, exc_info=stacktrace, **kwargs)
if more:
self._logger_log(ERROR, more)
return
# Default PdsLogger handling
exception_level = self._level_by_name['exception']
self.log('exception', message, *args, filepath=filepath, force=True)
if more:
self._logger_log(exception_level, more)
if stacktrace:
self._logger_log(exception_level, ''.join(traceback.format_tb(tb)))
[docs]
def blankline(self, level=INFO, force=False):
"""Write a blank line into the log.
Parameters:
level (int or str): Logging level or level name.
force (bool, optional): True to force the message to be logged even if the
logging level is above the specified `level`.
"""
if force:
level = CRITICAL
elif isinstance(level, str):
level = self._level_by_name[self._repair_level_name(level)]
self._logger_log(level, '')
def _logger_log(self, level, message, *args, **kwargs):
"""Log a message if it exceeds the logging level or is forced.
Parameters:
level (int): Logging level.
message (str): Complete message text.
*args: Additional arguments passed to self._logger.log().
*kwargs: Additional keyword arguments passed to self._logger.log().
"""
# Determine if any handlers are defined for this PdsLogger or its ancestry
has_handlers = False
plogger = self
while plogger is not None:
if plogger.handlers:
has_handlers = True
break
if not plogger.propagate:
break
plogger = plogger.parent
if not has_handlers: # if no handlers, print
print(self._format_message(level, message, *args, **kwargs))
else:
self._logger.log(level, message, *args, **kwargs)
######################################################################################
# Message formatting utilities
######################################################################################
def _logged_text(self, level, message, filepath='', *, shift=0):
"""Construct a record to send to the logger, including time tag, level indicator,
etc., in the standardized format.
Parameters:
level (int, str): Logging level or level name to appear in the log record.
message (str): Message text.
filepath (str, Path, or FCPath, optional): File path to include in the
message.
shift (int, optional):
Number of characters by which to shift the indent.
Returns:
str: The full text of the log message.
"""
parts = []
if self._timestamps:
timetag = datetime.datetime.now().strftime(_TIME_FMT)
if self._digits <= 0:
timetag = timetag[:19]
else:
timetag = timetag[:20+self._digits]
parts += [timetag, ' | ']
if self._lognames:
parts += [self._logname, ' | ']
if self._pid:
parts += [str(self._pid), ' | ']
if self._indent:
if parts:
parts[-1] = ' |'
dashes = max(0, self._get_depth() + shift)
parts += [dashes * '-', '| ']
if self._levelnames:
parts += [self._logged_level_name(level), ' | ']
parts.append(message)
filepath = self._logged_filepath(filepath)
if filepath:
parts += [': ', str(filepath)]
return ''.join(parts)
def _logged_filepath(self, filepath=''):
"""A file path to log, with any of the leading root paths stripped.
Parameters:
filepath (str, Path, or FCPath, optional): File path to append to the message.
Returns:
str: Path string to include in the logged message.
"""
if isinstance(filepath, (Path, FCPath)):
filepath = str(filepath)
if filepath == '.': # the result of Path('')
filepath = ''
if not filepath:
return ''
abspath = str(Path(filepath).absolute().resolve())
for root in self._roots:
if filepath.startswith(root) and filepath != root:
return filepath[len(root):]
if abspath.startswith(root) and abspath != root: # pragma: no cover
return abspath[len(root):]
return filepath
def _logged_level_name(self, level):
"""The name for a level to appear in the log, always upper case.
Parameters:
level (int, str): Logging level or level name.
Returns:
str: Level name to appear in the log.
"""
if isinstance(level, str):
return self._repair_level_name(level).upper()
level_name = self._level_names.get(level, '')
if level_name:
return level_name.upper()
# Use "<name>+i" where i is the smallest difference above a default name
diffs = [(level-lev, name.upper()) for lev, name in _DEFAULT_LEVEL_NAMES.items()]
diffs = [diff for diff in diffs if diff[0] > 0]
diffs.sort()
return f'{diffs[0][1]}+{diffs[0][0]}'
def _get_depth(self):
"""The current tier number (0-5) in the hierarchy."""
return len(self._titles) - 1
def _format_message(self, level, message, *args, **kwargs):
"""The given message with any substitutions applied, possibly including those from
the LogRecord.
"""
if '%' not in message:
return message
if _message_uses_args(message):
return message % args
if _message_uses_logrecord(message):
# Add the required attributes for an internal logger when first needed
if not hasattr(self, '_logrecords'):
self._logrecords = [] # destination of LogRecords
self._logrecord_logger = logging.getLogger('LogRecord')
self._logrecord_logger.setLevel(1)
self._logrecord_logger.addHandler(_LogRecordHandler(self._logrecords))
# Each message to this logger appends the LogRecord to self._logrecords
self._logrecord_logger.log(level, message)
logrecord = self._logrecords.pop()
kwargs.update(logrecord.__dict__)
kwargs['name'] = self.name # otherwise, it's "LogRecord"
# Add the asctime value to the logrecord dictionary if needed
if '%(asctime)' in message:
# Get a user-defined formatter if any
if (self._logger and self._logger.handlers
and self._logger.handlers[0].formatter): # pragma: no branch
formatter = self._logger.handlers[0].formatter # pragma: no cover
else:
formatter = logging._defaultFormatter
asctime = formatter.formatTime(logrecord)
kwargs['asctime'] = asctime
return message % kwargs
def _repair_level_name(self, name):
"""Replace the alias of a level with its default name."""
name = name.lower()
name = self._level_name_aliases.get(name, name)
return name
##########################################################################################
# Alternative loggers
##########################################################################################
[docs]
class EasyLogger(PdsLogger):
"""Simple subclass of PdsLogger that prints messages to the terminal."""
_LOGGER_IS_FAKE = True # Prevent registration as an actual logger
_NO_HANDLERS = True
[docs]
def __init__(self, logname='easylog', **kwargs):
PdsLogger.__init__(self, logname, **kwargs)
def _logger_log(self, level, message, *args, **kwargs):
"""Log a message if it exceeds the logging level or is forced.
Parameters:
level (int): Logging level.
message (str): Complete message text.
"""
print(self._format_message(level, message, *args, **kwargs))
[docs]
class ErrorLogger(EasyLogger):
"""Simple subclass of PdsLogger that suppresses all messages except ERROR messages and
those that have been forced.
"""
[docs]
def __init__(self, logname='errorlog', **kwargs):
PdsLogger.__init__(self, logname, **kwargs)
self.set_level(ERROR)
self._NO_LEVELS = True
[docs]
class CriticalLogger(EasyLogger):
"""Simple subclass of PdsLogger that suppresses all messages except CRITICAL messages
and those that have been forced.
"""
[docs]
def __init__(self, logname='criticallog', **kwargs):
PdsLogger.__init__(self, logname, **kwargs)
self.set_level(CRITICAL)
self._NO_LEVELS = True # Prevent changing the level
[docs]
class NullLogger(EasyLogger):
"""Simple subclass of PdsLogger that suppresses all messages."""
[docs]
def __init__(self, logname='nulllog', **kwargs):
PdsLogger.__init__(self, logname, **kwargs)
def _logger_log(self, level, message, *args, **kwargs):
return
##########################################################################################
# Support for the LogRecord formatting options
##########################################################################################
# An arg gets used by "%" not followed by "("
_FORMAT_ARG_FINDER = re.compile(r'%[^\(]')
def _message_uses_args(message):
"""True if the message contains substitutions to be used by args (not kwargs)."""
message = message.replace('%%', '') # in old-style formatting, ignore "%%"
return _FORMAT_ARG_FINDER.search(message) is not None
_ATTRIBUTE_FINDER = re.compile(r'%\((\w+)\)')
_ATTRIBUTE_NAMES = {'asctime', 'created', 'filename', 'funcName', 'levelname', 'levelno',
'lineno', 'message', 'module', 'msecs', 'name', 'pathname', 'process',
'processName', 'relativeCreated', 'thread', 'threadName', 'taskName'}
def _message_uses_logrecord(message):
"""True if the given message refers to an entry in the LogRecord."""
names = _ATTRIBUTE_FINDER.findall(message)
return bool(set(names) & _ATTRIBUTE_NAMES)
# https://stackoverflow.com/questions/57420008/python-after-logging-debug-how-to-view-
# its-logrecord
class _LogRecordHandler(logging.Handler):
"""A handler class that appends the latest LogRecord to an internal list."""
def __init__(self, records_list):
self.records_list = records_list
super().__init__()
def emit(self, record):
self.records_list.append(record)
##########################################################################################
# LoggerError
##########################################################################################
[docs]
class LoggerError(Exception):
"""For an exception of this class, `logger.exception()` will write a message into the
log with the user-specified level.
Unless requested, no stacktrace will be included in the log.
"""
[docs]
def __init__(self, message, filepath='', *, force=False, level='error',
stacktrace=False):
"""Constructor.
Parameters:
message (str or Exception):
Text of the message as a string. If an Exception object is provided, the
message is derived from this error.
filepath (str, Path, or FCPath, optional):
File path to include in the message. If not specified, the `filepath`
appearing in the call to `exception()` will be used.
force (bool, optional):
True to force the message to be logged even if the logging level is above
the level of "warning".
level (int or str, optional):
The level or level name for a record to enter the log.
stacktrace (bool, optional): True to include a stacktrace in the log.
"""
if isinstance(message, LoggerError):
self.__dict__ = message.__dict__
return
if isinstance(message, Exception):
self.message = str(message)
self.type_name = type(message).__name__
else:
self.message = str(message)
self.type_name = ''
self.filepath = filepath
self.force = force
self.level = level
if stacktrace:
_, _, self.tb = sys.exc_info()
else:
self.tb = None
[docs]
def __str__(self):
if self.type_name:
message = self.type_name + '(' + self.message + ')'
else:
message = self.message
if self.filepath:
return message + ': ' + str(self.filepath)
return message
##########################################################################################
# Handlers
##########################################################################################
STDOUT_HANDLER = logging.StreamHandler(sys.stdout)
STDOUT_HANDLER.setLevel(HIDDEN + 1)
stdout_handler = STDOUT_HANDLER # deprecated name
NULL_HANDLER = logging.NullHandler()
[docs]
def file_handler(logpath, level=HIDDEN+1, rotation='none', suffix=''):
"""File handler for a PdsLogger.
Parameters:
logath (str, Path, or FCPath):
The path to the log file.
level (int or str):
The minimum logging level at which to log messages; either an int or one of
"critical", "error", "warning", "info", "debug", or "hidden".
rotation (str, optional):
Log file rotation method, one of:
* "none": No rotation; append to an existing log of the same name.
* "number": Move an existing log file to one of the same name with a version
number ("_v" followed by an integer suffix of at least three digits) before
the extension.
* "midnight": Each night at midnight, append the date to the log file name and
start a new log.
* "ymd": Append the current date in the form "_yyyy-mm-dd" to each log file
name (before the ".log" extension).
* "ymdhms": Append the current date and time in the form
"_yyyy-mm-ddThh-mm-ss" to each log file name (before the ".log" extension).
* "replace": Replace this new log with any pre-existing log of the same name.
suffix (str, optional):
Append this suffix string to the log file name after any date and before the
extension.
Returns:
logging.FileHandler: FileHandler with the specified properties.
Raises:
ValueError: Invalid `rotation`.
KeyError: Invalid `level` string.
"""
def unlink_carefully(path):
try:
path.unlink()
except (NotImplementedError, FileNotFoundError):
if path.get_local_path().exists():
path.get_local_path().unlink() # pragma: no cover
if rotation not in {'none', 'number', 'midnight', 'ymd', 'ymdhms', 'replace'}:
raise ValueError(f'Unrecognized rotation for log file {logpath}: "{rotation}"')
if isinstance(level, str):
level = level.lower()
level = _DEFAULT_LEVEL_NAME_ALIASES.get(level, level)
level = _DEFAULT_LEVEL_BY_NAME[level]
# Define the FCPath and the local path
logpath = FCPath(logpath)
if not logpath.suffix:
logpath = logpath.with_suffix('.log')
local_logpath = logpath.get_local_path()
# Rename the previous log if rotation is "number"
if rotation == 'number':
if local_logpath.exists():
# Rename an existing log to one greater than the maximum numbered version
max_version = 0
regex = re.compile(logpath.stem + r'_v([0-9]+)' + logpath.suffix)
try:
for filepath in logpath.parent.glob(logpath.stem + '_v*'
+ logpath.suffix):
match = regex.match(filepath.name)
if match:
max_version = max(int(match.group(1)), max_version)
except NotImplementedError:
scheme = str(logpath).partition('://')[0]
raise ValueError('numbered rotation is not supported for remote scheme '
f'"{scheme}:"')
# Rename the existing log to one with a version number
basename = logpath.stem + '_v%03d' % (max_version+1) + logpath.suffix
versioned_logpath = logpath.parent / basename
versioned_local_logpath = local_logpath.parent / basename
os.rename(local_logpath, versioned_local_logpath)
try:
versioned_logpath.upload()
except NotImplementedError: # pragma: no cover
scheme = str(logpath).partition('://')[0]
raise ValueError('numbered rotation is not supported for remote scheme '
f'"{scheme}:"')
# Delete the existing file
unlink_carefully(logpath)
# Delete the previous log if rotation is 'replace'
elif rotation == 'replace':
unlink_carefully(logpath)
# Construct a dated log file name
elif rotation == 'ymd':
timetag = datetime.datetime.now().strftime('%Y-%m-%d')
logpath = logpath.with_stem(logpath.stem + '_' + timetag)
elif rotation == 'ymdhms':
timetag = datetime.datetime.now().strftime('%Y-%m-%dT%H-%M-%S')
logpath = logpath.with_stem(logpath.stem + '_' + timetag)
if suffix:
basename = logpath.stem + '_' + suffix.lstrip('_') + logpath.suffix
logpath = logpath.parent / basename
# Re-determine the local logpath
try:
local_logpath = logpath.retrieve()
except FileNotFoundError:
local_logpath = logpath.get_local_path()
# Create the handler
if rotation == 'midnight':
if not logpath.is_local():
raise ValueError('midnight rotation is not supported for remote files')
handler = logging.handlers.TimedRotatingFileHandler(
local_logpath,
when='midnight') # pragma: no cover
def _rotator(source, dest): # pragma: no cover
# This hack is required because the Python logging module is not
# multi-processor safe, and if there are multiple processes using the same log
# file for time rotation, they will all try to rename the file at midnight,
# but most will crash and burn because the log file is gone.
# Furthermore, we have to rename the destination log filename to something the
# logging module isn't expecting so that it doesn't later try to remove it in
# another process.
# See logging/handlers.py:392 (in Python 3.8)
try:
os.rename(source, dest + '_')
except FileNotFoundError:
pass
handler.rotator = _rotator # pragma: no cover
else:
handler = logging.FileHandler(local_logpath, mode='a')
handler.fcpath = logpath # save the FCPath as an extra attribute
handler.setLevel(level)
return handler
[docs]
def info_handler(parent, name='INFO.log', *, rotation='none'):
"""Quick creation of an "info"-level file handler.
Parameters:
parent (str, Path, or FCPath):
Path to the parent directory.
name (str, optional):
Basename of the file handler.
rotation (str, optional):
Log file rotation method, one of:
* "none": No rotation; append to an existing log of the same name.
* "number": Move an existing log file to one of the same name with a version
number ("_v" followed by an integer suffix of at least three digits) before
the extension.
* "midnight": Each night at midnight, append the date to the log file name and
start a new log.
* "ymd": Append the current date in the form "_yyyy-mm-dd" to each log file
name (before the ".log" extension).
* "ymdhms": Append the current date and time in the form
"_yyyy-mm-ddThh-mm-ss" to each log file name (before the ".log" extension).
* "replace": Replace this new log with any pre-existing log of the same name.
Returns:
logging.FileHandler: FileHandler with the specified properties.
Raises:
ValueError: Invalid `rotation`.
"""
return file_handler(FCPath(parent) / name, level=INFO, rotation=rotation)
[docs]
def warning_handler(parent, name='WARNINGS.log', *, rotation='none'):
"""Quick creation of a "warning"-level file handler.
Parameters:
parent (str, Path, or FCPath):
Path to the parent directory.
name (str, optional):
Basename of the file handler.
rotation (str, optional):
Log file rotation method, one of:
* "none": No rotation; append to an existing log of the same name.
* "number": Move an existing log file to one of the same name with a version
number ("_v" followed by an integer suffix of at least three digits) before
the extension.
* "midnight": Each night at midnight, append the date to the log file name and
start a new log.
* "ymd": Append the current date in the form "_yyyy-mm-dd" to each log file
name (before the ".log" extension).
* "ymdhms": Append the current date and time in the form
"_yyyy-mm-ddThh-mm-ss" to each log file name (before the ".log" extension).
* "replace": Replace this new log with any pre-existing log of the same name.
Returns:
logging.FileHandler: FileHandler with the specified properties.
Raises:
ValueError: Invalid `rotation`.
"""
return file_handler(FCPath(parent) / name, level=WARNING, rotation=rotation)
[docs]
def error_handler(parent, name='ERRORS.log', *, rotation='none'):
"""Quick creation of an "error"-level file handler.
Parameters:
parent (str, Path, or FCPath):
Path to the parent directory.
name (str, optional):
Basename of the file handler.
rotation (str, optional):
Log file rotation method, one of:
* "none": No rotation; append to an existing log of the same name.
* "number": Move an existing log file to one of the same name with a version
number ("_v" followed by an integer suffix of at least three digits) before
the extension.
* "midnight": Each night at midnight, append the date to the log file name and
start a new log.
* "ymd": Append the current date in the form "_yyyy-mm-dd" to each log file
name (before the ".log" extension).
* "ymdhms": Append the current date and time in the form
"_yyyy-mm-ddThh-mm-ss" to each log file name (before the ".log" extension).
* "replace": Replace this new log with any pre-existing log of the same name.
Returns:
logging.FileHandler: FileHandler with the specified properties.
Raises:
ValueError: Invalid `rotation`.
"""
return file_handler(FCPath(parent) / name, level=ERROR, rotation=rotation)
[docs]
def stream_handler(level=HIDDEN+1, stream=sys.stdout):
"""Stream handler, e.g., for directing logged messages to the terminal.
Parameters:
level (int or str, optional):
The minimum logging level at which to log messages; either an int or one of
"critical", "error", "warning", "info", "debug", or "hidden".
stream (stream, optional):
An output stream, defaulting to the terminal via sys.stdout.
"""
if isinstance(level, str):
level = level.lower()
level = _DEFAULT_LEVEL_NAME_ALIASES.get(level, level)
level = _DEFAULT_LEVEL_BY_NAME[level]
handler = logging.StreamHandler(stream)
handler.setLevel(level)
return handler
##########################################################################################