Python logging 模块，RootLogger() 实例源码

def get_logger(self, level : int) -> RootLogger:
        formatter = logging.Formatter(fmt='%(levelname)s (%(threadName)-10s) :%(name)s: %(message)s '
                                  '(%(asctime)s; %(filename)s:%(lineno)d)',
                              datefmt="%Y-%m-%d %H:%M:%S")
        handlers = [
            logging.handlers.RotatingFileHandler(self.__filename,
                                                 encoding='utf8',
                                                 maxBytes=self.__max_bytes,
                                                 backupCount=3),
            logging.StreamHandler()
        ]
        self.__root_logger = logging.getLogger()
        self.__root_logger.setLevel(level)
        for handler in handlers:
            handler.setFormatter(formatter)
            handler.setLevel(level)
            self.__root_logger.addHandler(handler)

        return self.__root_logger

def __init__(self, url, method='POST', fallback_file=None, loglevel=LOG_DEBUG, servicename=__default_servicename__):
        import urlparse
        self.loglevel = loglevel
        self.fallback_file = fallback_file
        p = urlparse.urlparse(url)
        netloc = p.netloc
        urlpath = p.path
        if p.query:
            urlpath += '?' + query
        h = HTTPHandler(host=netloc, url=urlpath, method=method)
        h.setLevel(loglevel)
        fmt = logging.Formatter(servicename+"[%(process)d]: %(levelname)s: %(message)s")
        h.setFormatter(fmt)
        self._logger = RootLogger(loglevel)
        self._logger.addHandler(h)
        if self.fallback_file:
            self._fallback = FileLogger(logfile=self.fallback_file,
                                        loglevel=self.loglevel,
                                        servicename=servicename)
        else:
            self._fallback = DummyLogger()

def root_logger(self) -> RootLogger:
        logging_config = LoggingConfig(general.logging['log_file'], general.logging['log_entries'])
        sys.excepthook = logging_config.set_error_hadler

        return logging_config.get_logger(logging.INFO)

def __init__(self, change_actuator_request_event: ChangeActuatorRequestEvent,
                 text_communication_enhancer: TextCommunicationEnhancer,
                 sound_api : SoundApi, logging: RootLogger):
        self.__change_actuator_request_event = change_actuator_request_event
        self.__text_communication_enhancer = text_communication_enhancer
        self.__sound_api = sound_api
        self.__logging = logging

def __init__(self, ifttt_rules: IftttRulesRepository, command_executor: CommandExecutor,
                 tokenizer: Tokenizer, logging: RootLogger):
        threading.Thread.__init__(self)
        self.__ifttt_rules = ifttt_rules
        self.__command_executor = command_executor
        self.__tokenizer = tokenizer
        self.__logging = logging
        self.shutdown = False

def __init__(self, root_logger: RootLogger) -> None:
        self.__root_logger = root_logger
        self.__token_converters = []

def __init__(self, tokenizer: Tokenizer, logger: RootLogger):
        self.__tokenizer = tokenizer
        self.__current_token_index = 0
        self.__logger = logger

def __init__(self, async_jobs: AsyncActuatorCommands, change_actuator_request_event: ChangeActuatorRequestEvent,
                 logging: RootLogger):
        threading.Thread.__init__(self)
        self.__async_jobs = async_jobs
        self.__change_actuator_request_event = change_actuator_request_event
        self.__logging = logging
        self.shutdown = False

def __init__(self, sensor_update_event: SensorUpdateEvent, sensors_repository: SensorsRepository,
                 actuators_repository: ActuatorsRepository, zwave_device: ZWaveDevice, logger: RootLogger):
        threading.Thread.__init__(self)
        self.__sensor_update_event = sensor_update_event
        self.__sensors_repository = sensors_repository
        self.__actuators_repository = actuators_repository
        self.__zwave_device = zwave_device
        self.__logger = logger
        self.shutdown = False

def __init__(self, root_logger: RootLogger, port: str, openzwave_config_path: str) -> None:
        self.__root_logger = root_logger
        self.__port = port
        self.__openzwave_config_path = openzwave_config_path
        self.__network = None
        self.__state_change_callback = None

def __init__(self, configuration, logger: RootLogger):
        self.__configuration = configuration
        self.__logger = logger
        self.__communicators = {}

def set_logger(self, logger: RootLogger):
        self.__logger = logger

def get_logger(self) -> RootLogger:
        return self.__logger

def __init__(self, root_logger: RootLogger) -> None:
        self.__root_logger = root_logger
        self.__env = None

def test_logger(self):
        import livebridge.config
        livebridge.config.LOGFILE = "/dev/null"
        from livebridge.logger import logger
        assert type(logger) is logging.RootLogger
        assert type(logger.handlers[0]) is logging.StreamHandler
        assert type(logger.handlers[1]) is logging.FileHandler
        logger.handlers = []

def __init__(self, loglevel=LOG_DEBUG, servicename=__default_servicename__):
        self.loglevel = loglevel
        h = logging.StreamHandler()
        h.setLevel(loglevel)
        fmt = logging.Formatter("%(asctime)s "+servicename+"[%(process)d]: %(levelname)s: %(message)s")
        h.setFormatter(fmt)
        self._logger = RootLogger(loglevel)
        self._logger.addHandler(h)

def __init__(self, addr='/dev/log', syslogfacility="local0", \
                 loglevel=LOG_DEBUG, servicename=__default_servicename__):
        if ':' in addr:
            host, port = addr.split(':', 1)
            port = int(port)
            addr = (host, port)
        fac = Syslog.facility_names[syslogfacility]
        h = Syslog(address=addr, facility=fac)
        h.setLevel(loglevel)
        fmt = logging.Formatter(servicename+"[%(process)d]: %(levelname)s: %(message)s")
        h.setFormatter(fmt)
        self._logger = RootLogger(loglevel)
        self._logger.addHandler(h)

def __init__(self, logfile='/tmp/%s.log' % __default_servicename__, \
                 loglevel=LOG_DEBUG, servicename=__default_servicename__):
        h = logging.FileHandler(filename=logfile)
        h.setLevel(loglevel)
        fmt = logging.Formatter("%(asctime)s "+servicename+"[%(process)d]: %(levelname)s: %(message)s")
        h.setFormatter(fmt)
        self._logger = RootLogger(loglevel)
        self._logger.addHandler(h)

def main(connection, logger, content_view=None, all_views=False, keep=0, **kwargs):
    """
    :type keep: int or bool
    :type all_views: bool
    :type connection: KatelloConnection
    :type content_view: list
    :param logger: Logger object
    :type logger: logging.RootLogger
    """

    if not keep:
        keep = None

    if all_views:
        logger.debug('Going to clean all content views as instructed, keeping {} versions'.format(keep))
        logger.debug('Getting all content views')
        view_names = [view['name'] for view in connection.content_views]
        logger.debug('Found: {} to clean'.format(all_views))
    else:
        try:
            logger.debug('Checking if {} is defined in config file')
            view_names = [c.strip() for c in kwargs['config_obj'].get(content_view[0], 'views').split(',')]
            logger.debug('Yes, we are going to clean {}'.format(view_names))
        except NoSectionError:
            logger.debug('No, we are cleaning {}'.format(content_view))
            view_names = content_view

    for view_name in view_names:
        try:
            logger.debug('Getting info for {}'.format(view_name))
            view_dict = get_components(connection.content_views, ('name', view_name))
        except TypeError:
            logger.error('View %s not found!' % view_name)
            raise NotFoundError('View {} not found'.format(view_name))

        ids_to_remove = [version['id'] for version in view_dict['versions'] if not version['environment_ids']]

        try:
            ids_to_remove = sorted(ids_to_remove)[:-keep]
        except TypeError:
            pass

        logger.debug("We are going to remove the following versions id's: {}".format(ids_to_remove))

        for version_id in ids_to_remove:
            logging.info('Removing version_id %s' % version_id)
            connection.remove_view_version(version_id)

def _configureLogger(name, force=False):
    try:
        logger = _Loggers[name]
    except KeyError:
        # add unknown logger names
        logger = getLogger(name)
        _debug("Added unknown logger name '%s'" % name)

    # If not forcing, skip loggers that already have handlers installed
    if not force and logger.handlers:
        return

    global _LogLevels
    if not _LogLevels:
        setLogLevels(getParam('GCAM.LogLevel') or 'WARN')

    if name in _LogLevels:
        level = _LogLevels[name]
        _debug("Configuring %s, level=%s" % (name, level))
        logger.setLevel(level)
    else:
        parent = logger.parent
        # Handle case of user plugins, which aren't in pygcam package
        # but we'd like them to default to the 'pygcam' loglevel
        if isinstance(parent, logging.RootLogger):
            parent = getLogger(PKGNAME)

        logger.setLevel(parent.level)

    logger.propagate = False

    # flush and remove all handlers
    for handler in logger.handlers:

        if not isinstance(handler, logging.NullHandler):
            handler.flush()
        logger.removeHandler(handler)

    logConsole = getParamAsBoolean('GCAM.LogConsole')
    if logConsole:
        consoleFormat = getParam('GCAM.LogConsoleFormat')
        _addHandler(logger, consoleFormat)

    logFile = getParam('GCAM.LogFile')
    if logFile:
        fileFormat = getParam('GCAM.LogFileFormat')
        _addHandler(logger, fileFormat, logFile=logFile)

    if not logger.handlers:
        logger.addHandler(logger, logging.NullHandler())
        _debug("Added NullHandler to root logger")