Python logbook 模块，StderrHandler() 实例源码

def analyze(context=None, results=None):
    import matplotlib.pyplot as plt
    import logbook
    logbook.StderrHandler().push_application()
    log = logbook.Logger('Algorithm')

    fig = plt.figure()
    ax1 = fig.add_subplot(211)

    results.algorithm_period_return.plot(ax=ax1,color='blue',legend=u'????')
    ax1.set_ylabel(u'??')
    results.benchmark_period_return.plot(ax=ax1,color='red',legend=u'????')

    plt.show()

# capital_base is the base value of capital
#

def main():
    """Plugin entrypoint."""
    with logbook.StderrHandler().applicationbound():
        # Read request message from stdin
        data = sys.stdin.buffer.read()

        # Parse request
        request = plugin.CodeGeneratorRequest()
        request.ParseFromString(data)

        # Create response
        response = plugin.CodeGeneratorResponse()

        # Generate code
        generate(request, response)

        # Serialise response message
        output = response.SerializeToString()

        # Write to stdout
        sys.stdout.buffer.write(output)

def initialize_cli(options):
    from ._logger import set_log_level

    debug_format_str = (
        "[{record.level_name}] {record.channel} {record.func_name} "
        "({record.lineno}): {record.message}")
    if options.log_level == logbook.DEBUG:
        info_format_str = debug_format_str
    else:
        info_format_str = (
            "[{record.level_name}] {record.channel}: {record.message}")

    logbook.StderrHandler(
        level=logbook.DEBUG, format_string=debug_format_str
    ).push_application()
    logbook.StderrHandler(
        level=logbook.INFO, format_string=info_format_str
    ).push_application()

    set_log_level(options.log_level)
    spr.SubprocessRunner.is_save_history = True

    if options.is_output_stacktrace:
        spr.SubprocessRunner.is_output_stacktrace = (
            options.is_output_stacktrace)

def test_stderr(
            self, capsys, command, ignore_stderr_regexp, out_regexp, expected):
        import logbook
        import subprocrunner

        logbook.StderrHandler(
            level=logbook.DEBUG).push_application()
        subprocrunner.set_log_level(logbook.INFO)

        runner = SubprocessRunner(
            command, ignore_stderr_regexp=ignore_stderr_regexp)
        runner.run()

        assert is_null_string(runner.stdout.strip())
        assert is_not_null_string(runner.stderr.strip())

        out, err = capsys.readouterr()
        print("[sys stdout]\n{}\n".format(out))
        print("[sys stderr]\n{}\n".format(err))
        print("[proc stdout]\n{}\n".format(runner.stdout))
        print("[proc stderr]\n{}\n".format(runner.stderr))

        actual = out_regexp.search(err) is not None
        assert actual == expected

def initialize_log_handler(log_level):
    debug_format_str = (
        "[{record.level_name}] {record.channel} {record.func_name} "
        "({record.lineno}): {record.message}")
    if log_level == logbook.DEBUG:
        info_format_str = debug_format_str
    else:
        info_format_str = (
            "[{record.level_name}] {record.channel}: {record.message}")

    logbook.StderrHandler(
        level=logbook.DEBUG, format_string=debug_format_str
    ).push_application()
    logbook.StderrHandler(
        level=logbook.INFO, format_string=info_format_str
    ).push_application()

def logging_context(path=None, level=None):
    from logbook import StderrHandler, FileHandler
    from logbook.compat import redirected_logging
    with StderrHandler(level=level or 'INFO').applicationbound():
        if path:
            if not os.path.isdir(os.path.dirname(path)):
                os.makedirs(os.path.dirname(path))
            with FileHandler(path, bubble=True).applicationbound():
                with redirected_logging():
                    yield
        else:
            with redirected_logging():
                yield

def main(extension, strict_extensions, default_extension):
    """Top level catalyst entry point.
    """
    # install a logbook handler before performing any other operations
    logbook.StderrHandler().push_application()
    load_extensions(
        default_extension,
        extension,
        strict_extensions,
        os.environ,
    )

def cli(ctx, verbosity, colored):
    """
    Command Line Interface. Use --help for details.
    """
    if verbosity > 0:
        try:
            # noinspection PyPackageRequirements,PyUnresolvedReferences
            import logbook
            # noinspection PyUnresolvedReferences,PyPackageRequirements
            import logbook.more
        except ImportError:
            click.echo('Please install saltyrtc.server[logging] for logging support.',
                       err=True)
            ctx.exit(code=_ErrorCode.import_error)

        # Translate logging level
        level = _get_logging_level(verbosity)

        # Enable asyncio debug logging if verbosity is high enough
        # noinspection PyUnboundLocalVariable
        if level <= logbook.DEBUG:
            os.environ['PYTHONASYNCIODEBUG'] = '1'

        # Enable logging
        util.enable_logging(level=level, redirect_loggers={
            'asyncio': level,
            'websockets': level,
        })

        # Get handler class
        if colored:
            handler_class = logbook.more.ColorizedStderrHandler
        else:
            handler_class = logbook.StderrHandler

        # Set up logging handler
        handler = handler_class(level=level)
        handler.push_application()
        ctx.obj['logging_handler'] = handler

def server_factory(request, event_loop, server_permanent_keys):
    """
    Return a factory to create :class:`saltyrtc.Server` instances.
    """
    # Enable asyncio debug logging
    os.environ['PYTHONASYNCIODEBUG'] = '1'

    # Enable logging
    util.enable_logging(level=logbook.NOTICE, redirect_loggers={
        'asyncio': logbook.WARNING,
        'websockets': logbook.WARNING,
    })

    # Push handler
    logging_handler = logbook.StderrHandler()
    logging_handler.push_application()

    _server_instances = []

    def _server_factory(permanent_keys=None):
        if permanent_keys is None:
            permanent_keys = server_permanent_keys

        # Setup server
        port = unused_tcp_port()
        coroutine = serve(
            util.create_ssl_context(
                pytest.saltyrtc.cert, dh_params_file=pytest.saltyrtc.dh_params),
            permanent_keys,
            host=pytest.saltyrtc.ip,
            port=port,
            loop=event_loop,
            server_class=TestServer,
        )
        server_ = event_loop.run_until_complete(coroutine)
        # Inject timeout and address (little bit of a hack but meh...)
        server_.timeout = _get_timeout(request=request)
        server_.address = (pytest.saltyrtc.ip, port)

        _server_instances.append(server_)

        def fin():
            server_.close()
            event_loop.run_until_complete(server_.wait_closed())
            _server_instances.remove(server_)
            if len(_server_instances) == 0:
                logging_handler.pop_application()

        request.addfinalizer(fin)
        return server_
    return _server_factory