Host._on_notification() - Code Metrics - Inspection of "plugin host: hide irrelevant parts of request trac..." - neovim/python-client - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#211)

by Björn

created 2016-06-08 19:24 UTC

Host._on_notification() A

↳ Parent: Host

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	2
CRAP Score	7.9297

Importance

Changes	4
Bugs	1	Features	0

Metric	Value
cc	3
c	4
b	1
f	0
dl	0
loc	13
ccs	2
cts	11
cp	0.1818
crap	7.9297
rs	9.4285

"""Implements a Nvim host for python plugins."""
import functools
import imp
import inspect
import logging
import os
import os.path
import re

from traceback import format_exc

from . import script_host
from ..api import decode_if_bytes, walk
from ..compat import IS_PYTHON3, find_module
from ..msgpack_rpc import ErrorResponse
from ..util import format_exc_skip


__all__ = ('Host')

logger = logging.getLogger(__name__)
error, debug, info, warn = (logger.error, logger.debug, logger.info,
                            logger.warning,)


class Host(object):

    """Nvim host for python plugins.

    Takes care of loading/unloading plugins and routing msgpack-rpc
    requests/notifications to the appropriate handlers.
    """

    def __init__(self, nvim):
        """Set handlers for plugin_load/plugin_unload."""
        self.nvim = nvim
        self._specs = {}
        self._loaded = {}
        self._load_errors = {}
        self._notification_handlers = {}
        self._request_handlers = {
            'poll': lambda: 'ok',
            'specs': self._on_specs_request,
            'shutdown': self.shutdown
        }

        # Decode per default for Python3
        self._decode_default = IS_PYTHON3

    def _on_async_err(self, msg):
        self.nvim.err_write(msg, async=True)

    def start(self, plugins):
        """Start listening for msgpack-rpc requests and notifications."""
        self.nvim.run_loop(self._on_request,
                           self._on_notification,
                           lambda: self._load(plugins),
                           err_cb=self._on_async_err)

    def shutdown(self):
        """Shutdown the host."""
        self._unload()
        self.nvim.stop_loop()

    def _wrap_function(self, fn, sync, decode, nvim_bind, name, *args):
        if decode:
            args = walk(decode_if_bytes, args, decode)
        if nvim_bind is not None:
            args.insert(0, nvim_bind)
        try:
            return fn(*args)
        except Exception as err:

            if sync:
                msg = ("error caught in request handler '{} {}':\n{!r}\n{}\n"
                       .format(name, args, err, format_exc_skip(1, 5)))
                raise ErrorResponse(msg)
            else:
                msg = ("error caught in async handler '{} {}':\n{!r}\n{}\n"
                       .format(name, args, err, format_exc_skip(1, 5)))
                self._on_async_err(msg + "\n")
                raise

    def _on_request(self, name, args):
        """Handle a msgpack-rpc request."""
        if IS_PYTHON3:
            name = decode_if_bytes(name)
        handler = self._request_handlers.get(name, None)
        if not handler:
            msg = self._missing_handler_error(name, 'request')
            error(msg)
            raise ErrorResponse(msg)

        debug('calling request handler for "%s", args: "%s"', name, args)
        rv = handler(*args)
        debug("request handler for '%s %s' returns: %s", name, args, rv)
        return rv

    def _on_notification(self, name, args):
        """Handle a msgpack-rpc notification."""
        if IS_PYTHON3:
            name = decode_if_bytes(name)
        handler = self._notification_handlers.get(name, None)
        if not handler:
            msg = self._missing_handler_error(name, 'notification')
            error(msg)
            self._on_async_err(msg + "\n")
            return

        debug('calling notification handler for "%s", args: "%s"', name, args)
        handler(*args)

    def _missing_handler_error(self, name, kind):
        msg = 'no {} handler registered for "{}"'.format(kind, name)
        pathmatch = re.match(r'(.+):[^:]+:[^:]+', name)
        if pathmatch:
            loader_error = self._load_errors.get(pathmatch.group(1))
            if loader_error is not None:
                msg = msg + "\n" + loader_error
        return msg

    def _load(self, plugins):
        for path in plugins:
            err = None
            if path in self._loaded:
                error('{0} is already loaded'.format(path))
                continue
            try:
                if path == "script_host.py":
                    module = script_host
                else:
                    directory, name = os.path.split(os.path.splitext(path)[0])
                    file, pathname, descr = find_module(name, [directory])
                    module = imp.load_module(name, file, pathname, descr)
                handlers = []
                self._discover_classes(module, handlers, path)
                self._discover_functions(module, handlers, path)
                if not handlers:
                    error('{0} exports no handlers'.format(path))
                    continue
                self._loaded[path] = {'handlers': handlers, 'module': module}
            except Exception as e:

                err = ('Encountered {} loading plugin at {}: {}\n{}'
                       .format(type(e).__name__, path, e, format_exc(5)))
                error(err)
                self._load_errors[path] = err

    def _unload(self):
        for path, plugin in self._loaded.items():

            handlers = plugin['handlers']
            for handler in handlers:
                method_name = handler._nvim_rpc_method_name
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                if hasattr(handler, '_nvim_shutdown_hook'):
                    handler()
                elif handler._nvim_rpc_sync:
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                    del self._request_handlers[method_name]
                else:
                    del self._notification_handlers[method_name]
        self._specs = {}
        self._loaded = {}

    def _discover_classes(self, module, handlers, plugin_path):
        for _, cls in inspect.getmembers(module, inspect.isclass):
            if getattr(cls, '_nvim_plugin', False):
                # create an instance of the plugin and pass the nvim object
                plugin = cls(self._configure_nvim_for(cls))
                # discover handlers in the plugin instance
                self._discover_functions(plugin, handlers, plugin_path)

    def _discover_functions(self, obj, handlers, plugin_path):
        def predicate(o):
            return hasattr(o, '_nvim_rpc_method_name')

        specs = []
        objdecode = getattr(obj, '_nvim_decode', self._decode_default)
        for _, fn in inspect.getmembers(obj, predicate):
            sync = fn._nvim_rpc_sync
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            decode = getattr(fn, '_nvim_decode', objdecode)
            nvim_bind = None
            if fn._nvim_bind:
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                nvim_bind = self._configure_nvim_for(fn)

            method = fn._nvim_rpc_method_name
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            if fn._nvim_prefix_plugin_path:
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
                method = '{0}:{1}'.format(plugin_path, method)

            fn_wrapped = functools.partial(self._wrap_function, fn,
                                           sync, decode, nvim_bind, method)
            self._copy_attributes(fn, fn_wrapped)
            # register in the rpc handler dict
            if sync:
                if method in self._request_handlers:
                    raise Exception(('Request handler for "{0}" is ' +
                                    'already registered').format(method))
                self._request_handlers[method] = fn_wrapped
            else:
                if method in self._notification_handlers:
                    raise Exception(('Notification handler for "{0}" is ' +
                                    'already registered').format(method))
                self._notification_handlers[method] = fn_wrapped
            if hasattr(fn, '_nvim_rpc_spec'):
                specs.append(fn._nvim_rpc_spec)
class MyParent:
    def __init__(self):
        self._x = 1;
        self.y = 2;

class MyChild(MyParent):
    def some_method(self):
        return self._x    # Ok, since accessed from a child class

class AnotherClass:
    def some_method(self, instance_of_my_child):
        return instance_of_my_child._x   # Would be flagged as AnotherClass is not
                                         # a child class of MyParent
            handlers.append(fn_wrapped)
        if specs:
            self._specs[plugin_path] = specs

    def _copy_attributes(self, fn, fn2):
class Foo:
    def some_method(self, x, y):
        return x + y;
        # Copy _nvim_* attributes from the original function
        for attr in dir(fn):
            if attr.startswith('_nvim_'):
                setattr(fn2, attr, getattr(fn, attr))

    def _on_specs_request(self, path):
        if IS_PYTHON3:
            path = decode_if_bytes(path)
        if path in self._load_errors:
            self.nvim.out_write(self._load_errors[path] + '\n')
        return self._specs.get(path, 0)

    def _configure_nvim_for(self, obj):
        # Configure a nvim instance for obj (checks encoding configuration)
        nvim = self.nvim
        decode = getattr(obj, '_nvim_decode', self._decode_default)
        if decode:
            nvim = nvim.with_decode(decode)
        return nvim


1		"""Implements a Nvim host for python plugins."""
2	6	import functools
3	6	import imp
4	6	import inspect
5	6	import logging
6	6	import os
7	6	import os.path
8	6	import re
9
10	6	from traceback import format_exc
11
12	6	from . import script_host
13	6	from ..api import decode_if_bytes, walk
14	6	from ..compat import IS_PYTHON3, find_module
15	6	from ..msgpack_rpc import ErrorResponse
16		from ..util import format_exc_skip
		0 ignored issues – show Bug introduced 2016-06-08 19:50 UTC by Report Bug Copy Issue Report The name `format_exc_skip` does not seem to exist in module `util`. Loading history...
17	6
18		__all__ = ('Host')
19	6
20	6	logger = logging.getLogger(__name__)
21		error, debug, info, warn = (logger.error, logger.debug, logger.info,
22		logger.warning,)
23
24	6
25		class Host(object):
26
27		"""Nvim host for python plugins.
28
29		Takes care of loading/unloading plugins and routing msgpack-rpc
30		requests/notifications to the appropriate handlers.
31		"""
32	6
33		def __init__(self, nvim):
34		"""Set handlers for plugin_load/plugin_unload."""
35		self.nvim = nvim
36		self._specs = {}
37		self._loaded = {}
38		self._load_errors = {}
39		self._notification_handlers = {}
40		self._request_handlers = {
41		'poll': lambda: 'ok',
42		'specs': self._on_specs_request,
43		'shutdown': self.shutdown
44		}
45
46		# Decode per default for Python3
47		self._decode_default = IS_PYTHON3
48	6
49		def _on_async_err(self, msg):
50		self.nvim.err_write(msg, async=True)
51	6
52		def start(self, plugins):
53		"""Start listening for msgpack-rpc requests and notifications."""
54		self.nvim.run_loop(self._on_request,
55		self._on_notification,
56		lambda: self._load(plugins),
57		err_cb=self._on_async_err)
58	6
59		def shutdown(self):
60		"""Shutdown the host."""
61		self._unload()
62		self.nvim.stop_loop()
63	6
64		def _wrap_function(self, fn, sync, decode, nvim_bind, name, *args):
65		if decode:
66		args = walk(decode_if_bytes, args, decode)
67		if nvim_bind is not None:
68		args.insert(0, nvim_bind)
69		try:
70		return fn(*args)
71		except Exception as err:
		0 ignored issues – show Best Practice introduced 2016-06-08 19:50 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
72		if sync:
73		msg = ("error caught in request handler '{} {}':\n{!r}\n{}\n"
74		.format(name, args, err, format_exc_skip(1, 5)))
75		raise ErrorResponse(msg)
76		else:
77		msg = ("error caught in async handler '{} {}':\n{!r}\n{}\n"
78	6	.format(name, args, err, format_exc_skip(1, 5)))
79		self._on_async_err(msg + "\n")
80		raise
81
82		def _on_request(self, name, args):
83		"""Handle a msgpack-rpc request."""
84		if IS_PYTHON3:
85		name = decode_if_bytes(name)
86		handler = self._request_handlers.get(name, None)
87		if not handler:
88		msg = self._missing_handler_error(name, 'request')
89		error(msg)
90		raise ErrorResponse(msg)
91
92		debug('calling request handler for "%s", args: "%s"', name, args)
93		rv = handler(*args)
94		debug("request handler for '%s %s' returns: %s", name, args, rv)
95		return rv
96
97		def _on_notification(self, name, args):
98	6	"""Handle a msgpack-rpc notification."""
99		if IS_PYTHON3:
100		name = decode_if_bytes(name)
101		handler = self._notification_handlers.get(name, None)
102		if not handler:
103		msg = self._missing_handler_error(name, 'notification')
104		error(msg)
105		self._on_async_err(msg + "\n")
106		return
107	6
108		debug('calling notification handler for "%s", args: "%s"', name, args)
109		handler(*args)
110
111		def _missing_handler_error(self, name, kind):
112		msg = 'no {} handler registered for "{}"'.format(kind, name)
113		pathmatch = re.match(r'(.+):[^:]+:[^:]+', name)
114		if pathmatch:
115		loader_error = self._load_errors.get(pathmatch.group(1))
116		if loader_error is not None:
117		msg = msg + "\n" + loader_error
118		return msg
119
120		def _load(self, plugins):
121		for path in plugins:
122		err = None
123		if path in self._loaded:
124		error('{0} is already loaded'.format(path))
125		continue
126		try:
127		if path == "script_host.py":
128		module = script_host
129		else:
130		directory, name = os.path.split(os.path.splitext(path)[0])
131		file, pathname, descr = find_module(name, [directory])
132		module = imp.load_module(name, file, pathname, descr)
133	6	handlers = []
134		self._discover_classes(module, handlers, path)
135		self._discover_functions(module, handlers, path)
136		if not handlers:
137		error('{0} exports no handlers'.format(path))
138		continue
139		self._loaded[path] = {'handlers': handlers, 'module': module}
140		except Exception as e:
		0 ignored issues – show Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report Catching very general exceptions such as `Exception` is usually not recommended. Generally, you would want to handle very specific errors in the exception handler. This ensure that you do not hide other types of errors which should be fixed. So, unless you specifically plan to handle any error, consider adding a more specific exception. Loading history...
141		err = ('Encountered {} loading plugin at {}: {}\n{}'
142		.format(type(e).__name__, path, e, format_exc(5)))
143		error(err)
144		self._load_errors[path] = err
145
146		def _unload(self):
147	6	for path, plugin in self._loaded.items():
		0 ignored issues – show Unused Code introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report The variable `path` seems to be unused. Loading history...
148		handlers = plugin['handlers']
149		for handler in handlers:
150		method_name = handler._nvim_rpc_method_name
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_method_name` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
151		if hasattr(handler, '_nvim_shutdown_hook'):
152		handler()
153		elif handler._nvim_rpc_sync:
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_sync` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
154		del self._request_handlers[method_name]
155	6	else:
156		del self._notification_handlers[method_name]
157		self._specs = {}
158		self._loaded = {}
159
160		def _discover_classes(self, module, handlers, plugin_path):
161		for _, cls in inspect.getmembers(module, inspect.isclass):
162		if getattr(cls, '_nvim_plugin', False):
163		# create an instance of the plugin and pass the nvim object
164		plugin = cls(self._configure_nvim_for(cls))
165		# discover handlers in the plugin instance
166		self._discover_functions(plugin, handlers, plugin_path)
167
168		def _discover_functions(self, obj, handlers, plugin_path):
169		def predicate(o):
170		return hasattr(o, '_nvim_rpc_method_name')
171
172		specs = []
173		objdecode = getattr(obj, '_nvim_decode', self._decode_default)
174		for _, fn in inspect.getmembers(obj, predicate):
175		sync = fn._nvim_rpc_sync
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_sync` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
176		decode = getattr(fn, '_nvim_decode', objdecode)
177		nvim_bind = None
178		if fn._nvim_bind:
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_bind` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
179		nvim_bind = self._configure_nvim_for(fn)
180
181		method = fn._nvim_rpc_method_name
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_method_name` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
182		if fn._nvim_prefix_plugin_path:
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_prefix_plugin_path` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
183		method = '{0}:{1}'.format(plugin_path, method)
184
185		fn_wrapped = functools.partial(self._wrap_function, fn,
186		sync, decode, nvim_bind, method)
187		self._copy_attributes(fn, fn_wrapped)
188		# register in the rpc handler dict
189		if sync:
190		if method in self._request_handlers:
191		raise Exception(('Request handler for "{0}" is ' +
192		'already registered').format(method))
193		self._request_handlers[method] = fn_wrapped
194		else:
195		if method in self._notification_handlers:
196	6	raise Exception(('Notification handler for "{0}" is ' +
197		'already registered').format(method))
198		self._notification_handlers[method] = fn_wrapped
199		if hasattr(fn, '_nvim_rpc_spec'):
200		specs.append(fn._nvim_rpc_spec)
		0 ignored issues – show Coding Style Best Practice introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report It seems like `_nvim_rpc_spec` was declared protected and should not be accessed from this context. Prefixing a member variable `_` is usually regarded as the equivalent of declaring it with protected visibility that exists in other languages. Consequentially, such a member should only be accessed from the same class or a child class: class MyParent: def __init__(self): self._x = 1; self.y = 2; class MyChild(MyParent): def some_method(self): return self._x # Ok, since accessed from a child class class AnotherClass: def some_method(self, instance_of_my_child): return instance_of_my_child._x # Would be flagged as AnotherClass is not # a child class of MyParent Loading history...
201		handlers.append(fn_wrapped)
202	6	if specs:
203		self._specs[plugin_path] = specs
204
205		def _copy_attributes(self, fn, fn2):
		0 ignored issues – show Coding Style introduced 2015-12-02 06:47 UTC by Report Bug Copy Issue Report This method could be written as a function/class method. If a method does not access any attributes of the class, it could also be implemented as a function or static method. This can help improve readability. For example class Foo: def some_method(self, x, y): return x + y; could be written as class Foo: @classmethod def some_method(cls, x, y): return x + y; Loading history...
206		# Copy _nvim_* attributes from the original function
207		for attr in dir(fn):
208		if attr.startswith('_nvim_'):
209	6	setattr(fn2, attr, getattr(fn, attr))
210
211		def _on_specs_request(self, path):
212		if IS_PYTHON3:
213		path = decode_if_bytes(path)
214		if path in self._load_errors:
215		self.nvim.out_write(self._load_errors[path] + '\n')
216		return self._specs.get(path, 0)
217
218		def _configure_nvim_for(self, obj):
219		# Configure a nvim instance for obj (checks encoding configuration)
220		nvim = self.nvim
221		decode = getattr(obj, '_nvim_decode', self._decode_default)
222		if decode:
223		nvim = nvim.with_decode(decode)
224		return nvim
225

neovim / python-client

Pull Request — master (#211)

Host._on_notification() A

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like