#!/usr/bin/python3
#
# Univention App Center
# appcenter logging module
#
# SPDX-FileCopyrightText: 2015-2025 Univention GmbH
# SPDX-License-Identifier: AGPL-3.0-only
#
"""
Univention App Center library:
Logging module
The library logs various messages to logger objects (python stdlib logging)
univention.appcenter.log defines the appcenter base logger, as well as
functions to link the logger objects to the application using the library.
>>> from univention.appcenter.log import *
>>> log_to_logfile()
>>> # logs all messages to '/var/log/univention/appcenter.log'
>>> log_to_stream()
>>> # logs messages other than debug to stdout or (warning/error) stderr
>>> base_logger = get_base_logger()
>>> base_logger.info('This is an info message')
>>> base_logger.warning('And this is a warning')
"""
import logging
import sys
from contextlib import contextmanager
LOG_FILE = '/var/log/univention/appcenter.log'
[docs]
def get_base_logger():
"""Returns the base logger for univention.appcenter"""
return logging.getLogger('univention.appcenter')
[docs]
class RangeFilter(logging.Filter):
"""
A Filter object that filters messages in a certain
range of logging levels
"""
def __init__(self, min_level=None, max_level=None):
super().__init__()
self.min_level = min_level
self.max_level = max_level
[docs]
def filter(self, record):
if self.max_level is None:
return record.levelno >= self.min_level
if self.min_level is None:
return record.levelno <= self.max_level
return self.min_level <= record.levelno <= self.max_level
[docs]
class UMCHandler(logging.Handler):
"""Handler to link a logger to the UMC logging mechanism"""
[docs]
def emit(self, record):
try:
from univention.management.console.log import MODULE
except ImportError:
pass
else:
msg = str(self.format(record))
if record.levelno <= logging.DEBUG:
MODULE.info(msg)
elif record.levelno <= logging.INFO:
MODULE.process(msg)
elif record.levelno <= logging.WARNING:
MODULE.warning(msg)
else:
MODULE.error(msg)
[docs]
class StreamReader:
def __init__(self, logger, level):
self.logger = logger
self.level = level
[docs]
def write(self, msg):
if self.logger:
self.logger.log(self.level, msg.rstrip('\n'))
[docs]
class LogCatcher:
def __init__(self, logger=None):
self._original_name = None
self.logger = logger
if logger:
self._original_name = logger.name
self.logs = []
[docs]
def getChild(self, name):
if self.logger:
self.logger.name = '%s.%s' % (self.logger.name, name)
return self
def __del__(self):
if self.logger and self._original_name:
self.logger.name = self._original_name
[docs]
def debug(self, msg, *args, **kwargs):
if self.logger:
self.logger.debug(msg, *args, **kwargs)
[docs]
def info(self, msg, *args, **kwargs):
if self.logger:
self.logger.info(msg, *args, **kwargs)
self.logs.append(('OUT', self._format(msg, args, **kwargs)))
[docs]
def warning(self, msg, *args, **kwargs):
if self.logger:
self.logger.warning(msg, *args, **kwargs)
self.logs.append(('ERR', self._format(msg, args, **kwargs)))
warn = warning
[docs]
def fatal(self, msg, *args, **kwargs):
if self.logger:
self.logger.warning(msg, *args, **kwargs)
self.logs.append(('ERR', self._format(msg, args, **kwargs)))
def _format(self, msg, args, exc_info=None, extra=None, stack_info=False, stacklevel=1):
log = self.logger or logging.getLogger("appcenter._internal")
record = log.makeRecord(
name=log.name,
level=logging.INFO,
fn="",
lno=0,
msg=msg,
args=args,
exc_info=exc_info,
func=None,
extra=extra,
sinfo=stack_info,
)
return record.getMessage()
[docs]
def has_stdout(self):
return any(self.stdout())
[docs]
def has_stderr(self):
return any(self.stderr())
[docs]
def stdout(self):
for level, msg in self.logs:
if level == 'OUT':
yield msg
[docs]
def stderr(self):
for level, msg in self.logs:
if level == 'ERR':
yield msg
[docs]
def stdstream(self):
for _level, msg in self.logs:
yield msg
def _reverse_umc_module_logger(exclusive=True):
"""
Function to redirect UMC logs to the univention.appcenter logger.
Useful when using legacy code when the App Center lib was part of the
UMC module
"""
try:
from univention.management.console.log import MODULE
except ImportError:
pass
else:
if exclusive:
for handler in MODULE.root.handlers:
MODULE.root.removeHandler(handler)
MODULE.parent = get_base_logger()
[docs]
@contextmanager
def catch_stdout(logger=None):
"""
Helper function to redirect stdout output to a logger. Or, if not
given, suppress completely. Useful when calling other libs that do not
use logging, instead just print statements.
"""
old_stdout = sys.stdout
old_stderr = sys.stderr
sys.stdout = StreamReader(logger, logging.INFO)
sys.stderr = StreamReader(logger, logging.WARNING)
try:
yield
finally:
sys.stdout = old_stdout
sys.stderr = old_stderr
[docs]
def log_to_stream():
"""
Call this function to log to stdout/stderr
stdout: logging.INFO
stderr: logging.WARNING and upwards
logging.DEBUG: suppressed
only the message is logged, no further formatting
stderr is logged in red (if its a tty)
"""
if not log_to_stream._already_set_up:
log_to_stream._already_set_up = True
logger = get_base_logger()
handler = logging.StreamHandler(sys.stdout)
handler.addFilter(RangeFilter(min_level=logging.INFO, max_level=logging.INFO))
logger.addHandler(handler)
handler = logging.StreamHandler(sys.stderr)
if sys.stderr.isatty():
formatter = logging.Formatter('\x1b[1;31m%(message)s\x1b[0m') # red
handler.setFormatter(formatter)
handler.addFilter(RangeFilter(min_level=logging.WARNING))
logger.addHandler(handler)
log_to_stream._already_set_up = False
[docs]
def get_logfile_logger(name):
mylogger = logging.getLogger(name)
mylogger.handlers = []
log_format = '%(process)6d %(short_name)-32s %(asctime)s [%(levelname)8s]: %(message)s'
log_format_time = '%y-%m-%d %H:%M:%S'
formatter = ShortNameFormatter(log_format, log_format_time)
handler = logging.FileHandler(LOG_FILE)
handler.setFormatter(formatter)
mylogger.addHandler(handler)
mylogger.setLevel(logging.DEBUG)
return mylogger
[docs]
def log_to_logfile():
"""
Call this function to log to /var/log/univention/appcenter.log
Needs rights to write to it (i.e. should be root)
Formats the message so that it can be analyzed later (i.e. process id)
Logs DEBUG as well
"""
if not log_to_logfile._already_set_up:
log_to_logfile._already_set_up = True
log_format = '%(process)6d %(short_name)-32s %(asctime)s [%(levelname)8s]: %(message)s'
log_format_time = '%y-%m-%d %H:%M:%S'
formatter = ShortNameFormatter(log_format, log_format_time)
handler = logging.FileHandler(LOG_FILE)
handler.setFormatter(formatter)
get_base_logger().addHandler(handler)
log_to_logfile._already_set_up = False
get_base_logger().setLevel(logging.DEBUG)
get_base_logger().addHandler(logging.NullHandler()) # this is to prevent warning messages
