console.py 文件源码

def execute_plural_server():  # pragma: no cover
    parser = argparse.ArgumentParser(
        prog='plural-server',
        description='ZMQ Reply Server that executes queries')

    parser.add_argument(
        '-b', '--reply-bind-addr',
        default='tcp://*:6000',
        help='ZMQ address to bind to'
    )

    parser.add_argument(
        '--color',
        action='store_true',
        default=True,
        help='colored logs'
    )
    parser.add_argument(
        '-l', '--log-level',
        default='DEBUG',
        choices=('DEBUG', 'INFO', 'WARNING')
    )

    args = parser.parse_args()
    server = GraphServer()
    if args.color:
        coloredlogs.install(level=args.log_level)

    logging.warning('EXPERIMENTAL FEATURE: server')
    try:
        server.run(args.reply_bind_addr)
    except KeyboardInterrupt:
        server.stop()
        sys.stdout.write("\r")
        sys.stderr.write("\r")
        sys.stdout.flush()
        sys.stderr.flush()