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

def main():
    """
    Creates all the argparse-rs and invokes the function from set_defaults().
    :return: The result of the function from set_defaults().
    """

    parser = get_parser()
    args = parser.parse_args()
    args.log_level = logging._nameToLevel[args.log_level]
    setup_logging(args.log_level)

    try:
        handler = args.handler
    except AttributeError:
        def print_usage(_):
            parser.print_usage()

        handler = print_usage
    return handler(args)

def __call__(self, parser, namespace, values, option_string=None):
        values = values.split(':')
        level, logger = values if len(values) > 1 else (values[0],
                                                        self.main_logger)

        logger = logging.getLogger(logger)
        try:
            logger.setLevel(logging._nameToLevel[level.upper()])
        except KeyError:
            msg = "invalid level choice: %s (choose from %s)" % \
                   (level, parser.log_levels)
            raise argparse.ArgumentError(self, msg)

        super(LoggingAction, self).__call__(parser, namespace, values,
                                            option_string)

def main():
    parser = argparse.ArgumentParser(
        description="asyncio-based source compiler and test runner")
    parser.add_argument(
        '-c',
        '--conf',
        type=argparse.FileType('r'),
        help="custom yaml configuration file to use")
    parser.add_argument(
        '-l',
        '--logging',
        choices=[l.lower() for l in logging._nameToLevel],
        help="logging level (overrides root logger level from file conf)")

    cmd = parser.add_subparsers(dest='command')
    commands = dict(getattr(module, 'build')(cmd)
                    for module in (languages, test, serve, benchmark))
    args = parser.parse_args()

    if args.conf:
        # merge user defined conf
        conf.merge(yaml.load(args.conf))

    # default logging config from conf
    logging.config.dictConfig(conf['logging'])

    if args.logging:
        # override root logger level
        logging.root.setLevel(args.logging.upper())

    # import built-in languages
    load_builtins()
    # import user languages from environ
    load_from_environ()

    try:
        func = commands[args.command]
    except KeyError:
        parser.error("missing command")
    else:
        sys.exit(func(args))

def setup_logging(level):
    """
    Makes stdout and stderr unicode friendly in case of misconfigured
    environments, initializes the logging and enables colored logs if it is
    appropriate.

    :param level: The logging level, can be either an int or a string.
    :return: None
    """
    if not isinstance(level, int):
        level = logging._nameToLevel[level]

    def ensure_utf8_stream(stream):
        if not isinstance(stream, io.StringIO) and hasattr(stream, "buffer"):
            stream = codecs.getwriter("utf-8")(stream.buffer)
            stream.encoding = "utf-8"
        return stream

    sys.stdout, sys.stderr = (ensure_utf8_stream(s)
                              for s in (sys.stdout, sys.stderr))
    logging.basicConfig(level=level)
    root = logging.getLogger()
    # In some cases root has handlers and basicConfig() is a no-op
    root.setLevel(level)
    if not sys.stdin.closed and sys.stdout.isatty():
        handler = root.handlers[0]
        handler.setFormatter(ColorFormatter())

def handle_logger(arglar):
    logger.setLevel(logging._nameToLevel[arglar.log_level])
    formatter = logging.Formatter('[%(asctime)s][%(name)s][%(levelname)s] %(message)s')
    logger.handlers = []
    if arglar.debug:
        handler = logging.handlers.RotatingFileHandler(filename=arglar.log_file.name,
                                                       maxBytes=1024 * 1024,
                                                       backupCount=3)
        handler.setFormatter(formatter)

        logger.addHandler(handler)
    else:
        logger.addHandler(logging.NullHandler())

def _process_level(value):
    """
    Given an input (as string or int) convert to LOGGING level value

    :param value:
    """

    # allow settings.ini to use names like level=debug, logging._nameToLevel[] will throw exception
    name_to_level = {
        'CRITICAL': logging.CRITICAL,
        'ERROR': logging.ERROR,
        'WARN': logging.WARNING,
        'WARNING': logging.WARNING,
        'INFO': logging.INFO,
        'DEBUG': logging.DEBUG,
        'NOTSET': logging.NOTSET
    }

    if isinstance(value, str):
        # handle the LOGGING level, such as INFO or DEBUG or 10
        value = unquote_string(value)
        try:
            # start assuming is int, but it is probably str of int
            value = int(value)

        except ValueError:
            # if here, then try if string like INFO, force to UPPER()
            try:
                value = name_to_level[value.upper()]

            except KeyError:
                raise ValueError("Logging level must as expected by Python LOGGING" +
                                 "module - not {}".format(value))

    if not isinstance(value, int):
        raise TypeError("Logging value must be string or int")

    return value

def set_msglevel(self, event):
        self.msg_level = logging._nameToLevel[self.cmb_msglevel.get().upper()]

def __getattr__(self, item):
        if item.upper() in logging._nameToLevel:
            return extras_wrapper(self, item)(getattr(self.logger, item))
        raise AttributeError

def main():
    """
    Creates all the argparse-rs and invokes the function from set_defaults().

    :return: The result of the function from set_defaults().
    """
    parser = argparse.ArgumentParser()
    parser.add_argument("--log-level", default="INFO",
                        choices=logging._nameToLevel,
                        help="Logging verbosity.")
    subparsers = parser.add_subparsers(help="Commands", dest="command")

    def add_backend_args(p):
        p.add_argument("--backend", default=None, help="Backend to use.")
        p.add_argument("--args", default=None, help="Backend's arguments.")

    dump_parser = subparsers.add_parser(
        "dump", help="Print a brief information about the model to stdout.")
    dump_parser.set_defaults(handler=dump_model)
    dump_parser.add_argument(
        "input", help="Path to the model file, URL or UUID.")
    add_backend_args(dump_parser)

    publish_parser = subparsers.add_parser(
        "publish", help="Upload the model to the cloud and update the "
                        "registry.")
    publish_parser.set_defaults(handler=publish_model)
    publish_parser.add_argument(
        "model", help="The path to the model to publish.")
    add_backend_args(publish_parser)
    publish_parser.add_argument("-d", "--update-default", action="store_true",
                                help="Set this model as the default one.")
    publish_parser.add_argument("--force", action="store_true",
                                help="Overwrite existing models.")

    list_parser = subparsers.add_parser(
        "list", help="Lists all the models in the registry.")
    list_parser.set_defaults(handler=list_models)
    add_backend_args(list_parser)

    init_parser = subparsers.add_parser("init", help="Initialize the registry.")
    init_parser.set_defaults(handler=initialize_registry)
    add_backend_args(init_parser)

    args = parser.parse_args()
    args.log_level = logging._nameToLevel[args.log_level]
    setup_logging(args.log_level)
    try:
        handler = args.handler
    except AttributeError:
        def print_usage(_):
            parser.print_usage()

        handler = print_usage
    return handler(args)

def logger():
    getpid_patch = patch('logging.os.getpid', return_value=111)
    getpid_patch.start()
    time_patch = patch('logging.time.time', return_value=946725071.111111)
    time_patch.start()
    localzone_patch = patch('rfc5424logging.handler.get_localzone', return_value=timezone)
    localzone_patch.start()
    hostname_patch = patch('rfc5424logging.handler.socket.gethostname', return_value="testhostname")
    hostname_patch.start()
    connect_patch = patch('logging.handlers.socket.socket.connect', side_effect=connect_mock)
    connect_patch.start()
    sendall_patch = patch('logging.handlers.socket.socket.sendall', side_effect=connect_mock)
    sendall_patch.start()

    if '_levelNames' in logging.__dict__:
        # Python 2.7
        level_patch = patch.dict(logging._levelNames)
        level_patch.start()
    else:
        # Python 3.x
        level_patch1 = patch.dict(logging._levelToName)
        level_patch1.start()
        level_patch2 = patch.dict(logging._nameToLevel)
        level_patch2.start()

    logging.raiseExceptions = True
    logger = logging.getLogger()
    logger.setLevel(logging.DEBUG)
    yield logger

    getpid_patch.stop()
    time_patch.stop()
    localzone_patch.stop()
    hostname_patch.stop()
    connect_patch.stop()
    sendall_patch.stop()

    if '_levelNames' in logging.__dict__:
        # Python 2.7
        level_patch.stop()
    else:
        # Python 3.x
        level_patch1.stop()
        level_patch2.stop()

    Rfc5424SysLogAdapter._extra_levels_enabled = False

def log(self):
        log = self.host.log
        if 'exc' in self.res:
            exc = '\n' + ''.join(self.res['exc'])
            log.error('{0}\n{1}'.format(self, exc))
        elif self.res.get('stderr'):
            if self.res.get('rc') != 0:
                log.error('{0}\n{1}'.format(self, self.res['stderr']))
            elif self.res['changed']:
                log.changed('{0}\n{1}'.format(self, self.res['stderr']))
        else:
            data = self.res.get('diff', '')
            if data.strip():
                data = data.strip()
                log.changed('{0} diff=\n{1}\n'.format(self, data))
            elif self.cancelled():
                log.error(self)
            elif self.res['changed']:
                log.changed(self)
            else:
                log.info(self)

        for cmd_ in self.meta.get('remote_calls', []):
            rtime = round(cmd_['time'], 3)
            inf = '^ sh({cmd}) time({rtime})'.format(
                rtime=rtime, **cmd_)
            if cmd_['exc']:
                log.error(inf + '\n' + ''.join(cmd_['exc']))
            elif nuka.cli.args.verbose > 1:
                log.info(inf)
                stds = {k: v for k, v in cmd_.items()
                        if k in ('stderr', 'stdout') and v}
                if stds:
                    log.debug3('^ ' + str(stds))

        data = self.res.get('log')
        if data:
            level = None
            for line in data.rstrip().split('\n'):
                line = line.rstrip()
                try:
                    line_level, message = line.split(':', 1)
                except ValueError:
                    if level:
                        log.log(level, '^ ' + line)
                else:
                    if line_level in logging._nameToLevel:
                        level = getattr(logging, line_level)
                        log.log(level, '^ ' + message)
                    else:
                        if level:
                            log.log(level, '^ ' + line)