Merge pull request #186 from onefinestay/cli

Cli

nameko/cli/actions.py

@@ -0,0 +1,30 @@
+import argparse
+import re
+
+
+class FlagAction(argparse.Action):
+    # From http://bugs.python.org/issue8538
+
+    def __init__(self, option_strings, dest, default=None,
+                 required=False, help=None, metavar=None,
+                 positive_prefixes=['--'], negative_prefixes=['--no-']):
+        self.positive_strings = set()
+        self.negative_strings = set()
+        for string in option_strings:
+            assert re.match(r'--[A-z]+', string)
+            suffix = string[2:]
+            for positive_prefix in positive_prefixes:
+                self.positive_strings.add(positive_prefix + suffix)
+            for negative_prefix in negative_prefixes:
+                self.negative_strings.add(negative_prefix + suffix)
+        strings = list(self.positive_strings | self.negative_strings)
+        super(FlagAction, self).__init__(
+            option_strings=strings, dest=dest,
+            nargs=0, const=None, default=default, type=bool, choices=None,
+            required=required, help=help, metavar=metavar)
+
+    def __call__(self, parser, namespace, values, option_string=None):
+        if option_string in self.positive_strings:
+            setattr(namespace, self.dest, True)
+        else:
+            setattr(namespace, self.dest, False)

nameko/cli/backdoor.py

@@ -0,0 +1,55 @@
+"""Connect to a nameko backdoor.
+
+If a backdoor is running this will connect to a remote shell.  The
+runner is generally available as `runner`.
+"""
+
+import os
+from subprocess import call
+
+from nameko.exceptions import CommandError
+from .actions import FlagAction
+
+
+def main(args):
+    for choice in ['netcat', 'nc', 'telnet']:
+        if os.system('which %s &> /dev/null' % choice) == 0:
+            prog = choice
+            break
+    else:
+        raise CommandError('Could not find an installed telnet.')
+
+    target = args.target
+    if ':' in target:
+        host, port = target.split(':', 1)
+    else:
+        host, port = 'localhost', target
+
+    rlwrap = args.rlwrap
+
+    cmd = [prog, str(host), str(port)]
+    if prog == 'netcat':
+        cmd.append('--close')
+    if rlwrap is None:
+        rlwrap = os.system('which rlwrap &> /dev/null') == 0
+    if rlwrap:
+        cmd.insert(0, 'rlwrap')
+    try:
+        if call(cmd) != 0:
+            raise CommandError(
+                'Backdoor unreachable on {}'.format(target)
+            )
+    except (EOFError, KeyboardInterrupt):
+        print
+        if choice == 'telnet' and rlwrap:
+            call(['reset'])
+
+
+def init_parser(parser):
+    parser.add_argument(
+        'target', metavar='[host:]port', help="(host and) port to connect to")
+    parser.add_argument(
+        '--rlwrap', dest='rlwrap', action=FlagAction,
+        help='Use rlwrap')
+    parser.set_defaults(feature=True)
+    return parser

nameko/cli/main.py

@@ -0,0 +1,26 @@
+import argparse
+
+from nameko.exceptions import CommandError
+from . import backdoor, run, shell
+
+
+def setup_parser():
+    parser = argparse.ArgumentParser()
+    subparsers = parser.add_subparsers()
+
+    for module in [backdoor, run, shell]:
+        name = module.__name__.split('.')[-1]
+        module_parser = subparsers.add_parser(
+            name, description=module.__doc__)
+        module.init_parser(module_parser)
+        module_parser.set_defaults(main=module.main)
+    return parser
+
+
+def main():
+    parser = setup_parser()
+    args = parser.parse_args()
+    try:
+        args.main(args)
+    except CommandError as exc:
+        print "Error: {}".format(exc)

nameko/cli/run.py

@@ -0,0 +1,155 @@
+"""Run a nameko service.
+
+Given a python path to a module containing a nameko service, will host and run
+it. By default this assumes a service class named ``Service``, but this can be
+provided via ``nameko run module:ServiceClass``.
+"""
+
+import eventlet
+
+eventlet.monkey_patch()
+
+
+import errno
+import logging
+import os
+import signal
+import sys
+
+from eventlet import backdoor
+
+from nameko.constants import AMQP_URI_CONFIG_KEY
+from nameko.exceptions import CommandError
+from nameko.runners import ServiceRunner
+
+
+logger = logging.getLogger(__name__)
+
+
+def import_service(module_name):
+    parts = module_name.split(":", 1)
+    if len(parts) == 1:
+        module_name, obj = module_name, "Service"
+    else:
+        module_name, obj = parts[0], parts[1]
+
+    try:
+        __import__(module_name)
+    except ImportError as exc:
+        if module_name.endswith(".py") and os.path.exists(module_name):
+            raise CommandError(
+                "Failed to find service, did you mean '{}'?".format(
+                    module_name[:-3].replace('/', '.')
+                )
+            )
+
+        missing_module_message = 'No module named {}'.format(module_name)
+        # is there a better way to do this?
+        if exc.message != missing_module_message:
+            # found module, but importing it raised an import error elsewhere
+            # let this bubble (resulting in a full stacktrace being printed)
+            raise
+
+        raise CommandError(exc)
+
+    module = sys.modules[module_name]
+
+    try:
+        service_cls = getattr(module, obj)
+    except AttributeError:
+        raise CommandError(
+            "Failed to find service class {!r} in module {!r}".format(
+                obj, module_name)
+            )
+
+    if not isinstance(service_cls, type):
+        raise CommandError("Service must be a class.")
+    return service_cls
+
+
+def setup_backdoor(runner, port):
+    def _bad_call():
+        raise RuntimeError(
+            'This would kill your service, not close the backdoor. To exit, '
+            'use ctrl-c.')
+    socket = eventlet.listen(('localhost', port))
+    gt = eventlet.spawn(
+        backdoor.backdoor_server,
+        socket,
+        locals={
+            'runner': runner,
+            'quit': _bad_call,
+            'exit': _bad_call,
+        })
+    return socket, gt
+
+
+def run(services, config, backdoor_port=None):
+    service_runner = ServiceRunner(config)
+    for service_cls in services:
+        service_runner.add_service(service_cls)
+
+    def shutdown(signum, frame):
+        # signal handlers are run by the MAINLOOP and cannot use eventlet
+        # primitives, so we have to call `stop` in a greenlet
+        eventlet.spawn_n(service_runner.stop)
+
+    signal.signal(signal.SIGTERM, shutdown)
+
+    if backdoor_port is not None:
+        setup_backdoor(service_runner, backdoor_port)
+
+    service_runner.start()
+
+    # if the signal handler fires while eventlet is waiting on a socket,
+    # the __main__ greenlet gets an OSError(4) "Interrupted system call".
+    # This is a side-effect of the eventlet hub mechanism. To protect nameko
+    # from seeing the exception, we wrap the runner.wait call in a greenlet
+    # spawned here, so that we can catch (and silence) the exception.
+    runnlet = eventlet.spawn(service_runner.wait)
+
+    while True:
+        try:
+            runnlet.wait()
+        except OSError as exc:
+            if exc.errno == errno.EINTR:
+                # this is the OSError(4) caused by the signalhandler.
+                # ignore and go back to waiting on the runner
+                continue
+            raise
+        except KeyboardInterrupt:
+            try:
+                service_runner.stop()
+            except KeyboardInterrupt:
+                service_runner.kill()
+        else:
+            # runner.wait completed
+            break
+
+
+def main(args):
+    logging.basicConfig(level=logging.INFO)
+
+    if '.' not in sys.path:
+        sys.path.insert(0, '.')
+
+    cls = import_service(args.service)
+
+    config = {AMQP_URI_CONFIG_KEY: args.broker}
+    run([cls], config, backdoor_port=args.backdoor_port)
+
+
+def init_parser(parser):
+    parser.add_argument(
+        'service', metavar='module[:service class]',
+        help='python path to the service class to run')
+    parser.add_argument(
+        '--broker', default='amqp://guest:guest@localhost:5672/nameko',
+        help='RabbitMQ broker url')
+    parser.add_argument(
+        '--backdoor-port',  type=int,
+        help='Specity a port number to host a backdoor, which can be connected'
+        ' to for an interactive interpreter within the running service'
+        ' process using `nameko backdoor`.'
+    )
+    return parser

nameko/cli/shell.py

@@ -0,0 +1,63 @@
+"""Launch an interactive python shell for working with remote nameko services.
+
+This is a regular interactive interpreter, with a special module ``n`` added
+to the builtin namespace, providing ``n.rpc`` and ``n.dispatch_event``.
+"""
+import code
+import os
+import sys
+from types import ModuleType
+
+from nameko.constants import AMQP_URI_CONFIG_KEY
+from nameko.standalone.rpc import ClusterRpcProxy
+from nameko.standalone.events import event_dispatcher
+
+
+def init_parser(parser):
+    parser.add_argument(
+        '--broker', default='amqp://guest:guest@localhost:5672/nameko',
+        help='RabbitMQ broker url')
+    return parser
+
+
+def make_nameko_helper(config):
+    """Create a fake module that provides some convenient access to nameko
+    standalone functionality for interactive shell usage.
+    """
+    module = ModuleType('nameko')
+    module.__doc__ = """Nameko shell helper for making rpc calls and dispaching
+events.
+
+Usage:
+    >>> n.rpc.service.method()
+    "reply"
+
+    >>> n.dispatch_event('service', 'event_type', 'event_data')
+"""
+    proxy = ClusterRpcProxy(config)
+    module.rpc = proxy.start()
+    module.dispatch_event = event_dispatcher(config)
+    module.config = config
+    module.disconnect = proxy.stop
+    return module
+
+
+def main(args):
+    banner = 'Nameko Python %s shell on %s\nBroker: %s' % (
+        sys.version,
+        sys.platform,
+        args.broker.encode('utf-8'),
+    )
+    config = {AMQP_URI_CONFIG_KEY: args.broker}
+
+    ctx = {}
+    ctx['n'] = make_nameko_helper(config)
+
+    # Support the regular Python interpreter startup script if someone
+    # is using it.
+    startup = os.environ.get('PYTHONSTARTUP')
+    if startup and os.path.isfile(startup):
+        with open(startup, 'r') as f:
+            eval(compile(f.read(), startup, 'exec'), ctx)
+
+    code.interact(banner=banner, local=ctx)

nameko/constants.py

@@ -1,3 +1,5 @@
+AMQP_URI_CONFIG_KEY = 'AMQP_URI'
+
 CALL_ID_STACK_CONTEXT_KEY = 'call_id_stack'
 AUTH_TOKEN_CONTEXT_KEY = 'auth_token'
 LANGUAGE_CONTEXT_KEY = 'language'

nameko/exceptions.py

@@ -132,3 +132,7 @@ def __init__(self, value):
 
     def __str__(self):
         return "Unserializable value: `{}`".format(self.repr_value)
+
+
+class CommandError(Exception):
+    """Raise from subcommands to report error back to the user"""

nameko/legacy/dependencies.py

@@ -5,12 +5,11 @@
 from kombu import Connection
 from kombu.pools import producers
 
-from nameko.constants import DEFAULT_RETRY_POLICY
+from nameko.constants import DEFAULT_RETRY_POLICY, AMQP_URI_CONFIG_KEY
 from nameko.exceptions import ContainerBeingKilled
 from nameko.dependencies import (
     dependency, entrypoint, DependencyFactory, CONTAINER_SHARED)
 from nameko.legacy.nova import get_topic_queue, parse_message
-from nameko.messaging import AMQP_URI_CONFIG_KEY
 from nameko.rpc import RpcConsumer, RpcProvider, Responder
 
 

nameko/messaging.py

@@ -15,7 +15,7 @@
 from kombu import Connection
 from kombu.mixins import ConsumerMixin
 
-from nameko.constants import DEFAULT_RETRY_POLICY
+from nameko.constants import DEFAULT_RETRY_POLICY, AMQP_URI_CONFIG_KEY
 from nameko.dependencies import (
     InjectionProvider, EntrypointProvider, entrypoint, injection,
     DependencyProvider, ProviderCollector, DependencyFactory, dependency,
@@ -27,7 +27,6 @@
 # delivery_mode
 PERSISTENT = 2
 HEADER_PREFIX = "nameko"
-AMQP_URI_CONFIG_KEY = 'AMQP_URI'
 
 
 class HeaderEncoder(object):

nameko/rpc.py

@@ -11,12 +11,11 @@
 from kombu import Connection, Exchange, Queue
 from kombu.pools import producers
 
-from nameko.constants import DEFAULT_RETRY_POLICY
+from nameko.constants import DEFAULT_RETRY_POLICY, AMQP_URI_CONFIG_KEY
 from nameko.exceptions import (
     MethodNotFound, UnknownService, UnserializableValueError,
     MalformedRequest, RpcConnectionError, serialize, deserialize)
-from nameko.messaging import (
-    queue_consumer, HeaderEncoder, HeaderDecoder, AMQP_URI_CONFIG_KEY)
+from nameko.messaging import queue_consumer, HeaderEncoder, HeaderDecoder
 from nameko.dependencies import (
     entrypoint, injection, InjectionProvider, EntrypointProvider,
     DependencyFactory, dependency, ProviderCollector, DependencyProvider,