Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
ret_path = logfile
if not ret_path and logfile and logdir:
ret_path = os.path.join(logdir, logfile)
if not ret_path and logdir:
program_name = program_name or get_program_name()
ret_path = os.path.join(logdir, program_name) + logfile_suffix
if not ret_path:
raise ValueError("Unable to determine log file destination")
return ret_path
class File(Output):
"""Ouput to a file."""
def __init__(self, filename=None, directory=None, suffix=".log",
program_name=None, formatter=formatter.TEXT_FORMATTER,
level=None):
"""Log file output.
:param filename: The log file path to write to.
If directory is also specified, both will be combined.
:param directory: The log directory to write to.
If no filename is specified, the program name and suffix will be used
to contruct the full path relative to the directory.
:param suffix: The log file name suffix.
This will be only used if no filename has been provided.
:param program_name: Program name. Autodetected by default.
"""
predetermined size.
:param backup_count: the maximum number of files to rotate
logging output between.
"""
logpath = _get_log_file_path(filename, directory,
program_name, suffix)
handler = logging.handlers.RotatingFileHandler(
logpath, maxBytes=max_size_bytes, backupCount=backup_count)
super(RotatingFile, self).__init__(handler, formatter, level)
def do_rollover(self):
"""Manually forces a log file rotation."""
return self.handler.doRollover()
class TimedRotatingFile(Output):
"""Rotating log file output, triggered by a fixed interval."""
def __init__(self, filename=None, directory=None, suffix='.log',
program_name=None, formatter=formatter.TEXT_FORMATTER,
level=None, interval=datetime.timedelta(hours=24),
backup_count=0):
"""Rotating log file output, triggered by a fixed interval.
:param filename: The log file path to write to.
If directory is also specified, both will be combined.
:param directory: The log directory to write to.
If no filename is specified, the program name and suffix will be used
to contruct the full path relative to the directory.
:param suffix: The log file name suffix.
This will be only used if no filename has been provided.
:param program_name: Program name. Autodetected by default.
@staticmethod
def _timedelta_to_seconds(td):
"""Convert a datetime.timedelta object into a seconds interval for
rotating file ouput.
:param td: datetime.timedelta
:return: time in seconds
:rtype: int
"""
if isinstance(td, numbers.Real):
td = datetime.timedelta(seconds=td)
return td.total_seconds()
class Stream(Output):
"""Generic stream output."""
def __init__(self, stream=sys.stderr, formatter=formatter.TEXT_FORMATTER,
level=None):
super(Stream, self).__init__(handlers.TTYDetectorStreamHandler(stream),
formatter, level)
STDERR = Stream()
STDOUT = Stream(sys.stdout)
class Journal(Output):
def __init__(self, program_name=None,
formatter=formatter.TEXT_FORMATTER, level=None):
program_name = program_name or get_program_name
:param filename: The log file path to write to.
If directory is also specified, both will be combined.
:param directory: The log directory to write to.
If no filename is specified, the program name and suffix will be used
to contruct the full path relative to the directory.
:param suffix: The log file name suffix.
This will be only used if no filename has been provided.
:param program_name: Program name. Autodetected by default.
"""
logpath = _get_log_file_path(filename, directory,
program_name, suffix)
handler = logging.handlers.WatchedFileHandler(logpath)
super(File, self).__init__(handler, formatter, level)
class RotatingFile(Output):
"""Output to a file, rotating after a certain size."""
def __init__(self, filename=None, directory=None, suffix='.log',
program_name=None, formatter=formatter.TEXT_FORMATTER,
level=None, max_size_bytes=0, backup_count=0):
"""Rotating log file output.
:param filename: The log file path to write to.
If directory is also specified, both will be combined.
:param directory: The log directory to write to.
If no filename is specified, the program name and suffix will be used
to contruct the full path relative to the directory.
:param suffix: The log file name suffix.
This will be only used if no filename has been provided.
:param program_name: Program name. Autodetected by default.
:param max_size_bytes: allow the file to rollover at a
formatter, level)
STDERR = Stream()
STDOUT = Stream(sys.stdout)
class Journal(Output):
def __init__(self, program_name=None,
formatter=formatter.TEXT_FORMATTER, level=None):
program_name = program_name or get_program_name
super(Journal, self).__init__(handlers.JournalHandler(program_name),
formatter, level)
class Syslog(Output):
def __init__(self, program_name=None, facility="user",
formatter=formatter.TEXT_FORMATTER, level=None):
if syslog is None:
# FIXME(jd) raise something more specific
raise RuntimeError("syslog is not available on this platform")
super(Syslog, self).__init__(
handlers.SyslogHandler(
program_name=program_name or get_program_name(),
facility=self._find_facility(facility)),
formatter, level)
@staticmethod
def _find_facility(facility):
# NOTE(jd): Check the validity of facilities at run time as they differ
# depending on the OS and Python version being used.
valid_facilities = [f for f in
class Stream(Output):
"""Generic stream output."""
def __init__(self, stream=sys.stderr, formatter=formatter.TEXT_FORMATTER,
level=None):
super(Stream, self).__init__(handlers.TTYDetectorStreamHandler(stream),
formatter, level)
STDERR = Stream()
STDOUT = Stream(sys.stdout)
class Journal(Output):
def __init__(self, program_name=None,
formatter=formatter.TEXT_FORMATTER, level=None):
program_name = program_name or get_program_name
super(Journal, self).__init__(handlers.JournalHandler(program_name),
formatter, level)
class Syslog(Output):
def __init__(self, program_name=None, facility="user",
formatter=formatter.TEXT_FORMATTER, level=None):
if syslog is None:
# FIXME(jd) raise something more specific
raise RuntimeError("syslog is not available on this platform")
super(Syslog, self).__init__(
handlers.SyslogHandler(
program_name=program_name or get_program_name(),