Source code for gnes.cli.api

def _start_service(cls, args):
    from ..service.base import ServiceManager
    with ServiceManager(cls, args) as es:

[docs]def grpc(args): from ..service.grpc import GRPCService _start_service(GRPCService, args)
[docs]def preprocess(args): from ..service.preprocessor import PreprocessorService _start_service(PreprocessorService, args)
[docs]def encode(args): from ..service.encoder import EncoderService _start_service(EncoderService, args)
[docs]def index(args): from ..service.indexer import IndexerService _start_service(IndexerService, args)
[docs]def route(args): from ..service.router import RouterService _start_service(RouterService, args)
[docs]def frontend(args): from ..service.frontend import FrontendService _start_service(FrontendService, args)
[docs]def client(args): if args.client == 'cli': return _client_cli(args) else: raise ValueError('gnes client must follow with a client type from {http, cli, benchmark...}\n' 'see "gnes client --help" for details')
[docs]def healthcheck(args): from ..service.base import send_ctrl_message from ..proto import gnes_pb2, add_version import time ctrl_addr = 'tcp://%s:%d' % (, args.port) msg = gnes_pb2.Message() add_version(msg.envelope) msg.request.control.command = gnes_pb2.Request.ControlRequest.STATUS for j in range(args.retries): r = send_ctrl_message(ctrl_addr, msg, timeout=args.timeout) if not r: print('%s is not responding, retry (%d/%d) in 1s' % (ctrl_addr, j + 1, args.retries)) else: print('%s returns %s' % (ctrl_addr, r)) exit(0) time.sleep(1) exit(1)
def _client_cli(args): from ..client.cli import CLIClient CLIClient(args)